[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_228":-5,"_231":232,"_233":234},"routes/shell",{"_40":41,"_227":-5,"_228":-5,"_229":230},"routes/repo",{"_33":-7,"_34":35,"_36":37,"_38":39,"_40":41,"_42":31,"_43":31,"_44":31,"_45":31,"_46":31,"_47":31,"_48":49,"_50":49,"_51":31,"_52":-7,"_53":-7,"_54":55,"_56":57,"_58":59,"_60":-5,"_61":62,"_63":64,"_65":66,"_67":68},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":14,"_22":23,"_24":-5,"_25":26,"_27":28,"_28":29,"_30":31,"_32":14},"ref","828bc0e5aba5912b8725bb529107b48bdf3ab259","path","apps/server/src/storage/postgres/adapter.ts","oid","84b34f84234bbefd0857d777254c7bbc462f0f58","size",36620,"commit","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","import { SQL } from \"bun\";\n\nimport { documentSlug, documentSlugCandidate } from \"../../channels/slug\";\nimport { conflict, corrupt, missing, StorageError, unavailable } from \"../errors\";\nimport { migrate, verifyMigrations } from \"./migrations\";\nimport { PostgresNavigationStore } from \"./navigation\";\nimport { PostgresBackgroundJobStore } from \"./jobs\";\n\nimport type { TransactionSQL } from \"bun\";\nimport type {\n\tAgentState,\n\tChannelCursor,\n\tChannelPage,\n\tChannelRecord,\n\tChannelScanCursor,\n\tChannelScanPage,\n\tChannelSnapshot,\n\tChannelUpdate,\n\tCommitChannel,\n\tCommitResult,\n\tCreateChannel,\n\tJsonValue,\n\tLease,\n\tRenameChannel,\n\tRenameResult,\n\tReplaceChannel,\n\tSaveCheckpoint,\n\tStoredChannel,\n\tStoredEvent,\n\tUpdateAgentContext,\n\tUserRecord,\n\tWebSession,\n} from \"../model\";\nimport type {\n\tBackgroundJobStore,\n\tChannelStore,\n\tCollaborationStore,\n\tLeaseStore,\n\tNavigationStore,\n\tSessionStore,\n\tStorageAdapter,\n\tUserStore,\n} from \"../port\";\n\ntype Timestamp = Date | string;\ntype Integer = bigint | number | string;\n\ntype UserRow = {\n\tid: string;\n\tlogin: string;\n\tavatarUrl: string;\n\tcreatedAt: Timestamp;\n\tupdatedAt: Timestamp;\n};\n\ntype SessionRow = {\n\tid: string;\n\tuserId: string;\n\texpiresAt: Timestamp;\n\tcreatedAt: Timestamp;\n};\n\ntype ChannelRow = {\n\tid: string;\n\trepositoryId: string;\n\trepositoryOwner: string;\n\trepositoryName: string;\n\ttitle: string;\n\tslug?: string;\n\tcreatedBy: string;\n\trevision: Integer;\n\tnextSequence?: Integer;\n\tcreatedAt: Timestamp;\n\tupdatedAt: Timestamp;\n};\n\ntype SnapshotRow = {\n\tchannelId: string;\n\tgeneration: string;\n\trevision: Integer;\n\tthroughSequence: Integer;\n\tepoch: string;\n\tsource: string;\n\tsourceHash: string;\n\tdocument: Uint8Array;\n\tsidecar: unknown;\n\tcreatedAt: Timestamp;\n};\n\ntype UpdateRow = {\n\tchannelId: string;\n\tsequence: Integer;\n\trevision: Integer;\n\tepoch: string;\n\tupdate: Uint8Array;\n};\n\ntype EventRow = {\n\tid: string;\n\tchannelId: string;\n\tsequence: Integer;\n\tordinal: number;\n\tkind: string;\n\tpayload: unknown;\n\tcreatedAt: Timestamp;\n};\n\ntype AgentRow = {\n\tchannelId: string;\n\townerSessionId: string | null;\n\tgeneration: Integer;\n\tsummary: string;\n\ttranscriptCursor: Integer;\n\tstatus: string;\n\tupdatedAt: Timestamp;\n};\n\ntype LeaseRow = {\n\tname: string;\n\towner: string;\n\tfencing: Integer;\n\texpiresAt: Timestamp;\n};\n\ntype OperationRow = {\n\trevision: Integer;\n\tsequence: Integer;\n};\n\ntype ChannelLockRow = {\n\trevision: Integer;\n\tnextSequence: Integer;\n};\n\nfunction date(value: Timestamp, field: string): Date {\n\tlet parsed = value instanceof Date ? new Date(value) : new Date(value);\n\tif (Number.isNaN(parsed.getTime())) throw corrupt(`storage returned an invalid ${field}`);\n\treturn parsed;\n}\n\nfunction integer(value: Integer, field: string): number {\n\tlet parsed = typeof value === \"number\" ? value : Number(value);\n\tif (!Number.isSafeInteger(parsed) || parsed < 0) {\n\t\tthrow corrupt(`storage returned an invalid ${field}`);\n\t}\n\treturn parsed;\n}\n\nfunction bytes(value: Uint8Array, field: string): Uint8Array {\n\tif (!(value instanceof Uint8Array)) throw corrupt(`storage returned an invalid ${field}`);\n\treturn new Uint8Array(value);\n}\n\nfunction validJson(value: unknown): value is JsonValue {\n\tif (value === null || typeof value === \"string\" || typeof value === \"boolean\") return true;\n\tif (typeof value === \"number\") return Number.isFinite(value);\n\tif (Array.isArray(value)) return value.every(validJson);\n\tif (typeof value !== \"object\") return false;\n\treturn Object.values(value as Record).every(validJson);\n}\n\nfunction json(value: unknown, field: string): JsonValue {\n\tlet parsed = typeof value === \"string\" ? parseJson(value, field) : value;\n\tif (!validJson(parsed)) throw corrupt(`storage returned invalid JSON for ${field}`);\n\treturn structuredClone(parsed);\n}\n\nfunction parseJson(value: string, field: string): unknown {\n\ttry {\n\t\treturn JSON.parse(value);\n\t} catch (err) {\n\t\tthrow corrupt(`storage returned invalid JSON for ${field}`, err);\n\t}\n}\n\nfunction user(row: UserRow): UserRecord {\n\treturn {\n\t\t...row,\n\t\tcreatedAt: date(row.createdAt, \"user creation time\"),\n\t\tupdatedAt: date(row.updatedAt, \"user update time\"),\n\t};\n}\n\nfunction session(row: SessionRow): WebSession {\n\treturn {\n\t\t...row,\n\t\texpiresAt: date(row.expiresAt, \"session expiry\"),\n\t\tcreatedAt: date(row.createdAt, \"session creation time\"),\n\t};\n}\n\nfunction channel(row: ChannelRow): ChannelRecord {\n\tif (!row.slug) throw corrupt(`channel ${row.id} has no canonical slug`);\n\treturn {\n\t\t...row,\n\t\tslug: row.slug,\n\t\trevision: integer(row.revision, \"channel revision\"),\n\t\tcreatedAt: date(row.createdAt, \"channel creation time\"),\n\t\tupdatedAt: date(row.updatedAt, \"channel update time\"),\n\t};\n}\n\nfunction snapshot(row: SnapshotRow): ChannelSnapshot {\n\treturn {\n\t\t...row,\n\t\trevision: integer(row.revision, \"snapshot revision\"),\n\t\tthroughSequence: integer(row.throughSequence, \"snapshot sequence\"),\n\t\tdocument: bytes(row.document, \"snapshot document\"),\n\t\tsidecar: json(row.sidecar, \"snapshot sidecar\"),\n\t\tcreatedAt: date(row.createdAt, \"snapshot creation time\"),\n\t};\n}\n\nfunction update(row: UpdateRow): ChannelUpdate {\n\treturn {\n\t\t...row,\n\t\tsequence: integer(row.sequence, \"update sequence\"),\n\t\trevision: integer(row.revision, \"update revision\"),\n\t\tupdate: bytes(row.update, \"Yjs update\"),\n\t};\n}\n\nfunction event(row: EventRow): StoredEvent {\n\treturn {\n\t\t...row,\n\t\tsequence: integer(row.sequence, \"event sequence\"),\n\t\tpayload: json(row.payload, \"event payload\"),\n\t\tcreatedAt: date(row.createdAt, \"event creation time\"),\n\t};\n}\n\nfunction agent(row: AgentRow): AgentState {\n\tif (row.status !== \"ready\" && row.status !== \"unavailable\") {\n\t\tthrow corrupt(\"storage returned an invalid agent status\");\n\t}\n\treturn {\n\t\t...row,\n\t\townerSessionId: row.ownerSessionId ?? undefined,\n\t\tgeneration: integer(row.generation, \"agent ownership generation\"),\n\t\ttranscriptCursor: integer(row.transcriptCursor, \"agent transcript cursor\"),\n\t\tstatus: row.status,\n\t\tupdatedAt: date(row.updatedAt, \"agent update time\"),\n\t};\n}\n\nfunction lease(row: LeaseRow): Lease {\n\treturn {\n\t\t...row,\n\t\tfencing: integer(row.fencing, \"lease fencing token\"),\n\t\texpiresAt: date(row.expiresAt, \"lease expiry\"),\n\t};\n}\n\nfunction postgresCode(err: unknown): string | undefined {\n\tif (!err || typeof err !== \"object\") return undefined;\n\tif (\"errno\" in err && typeof err.errno === \"string\") return err.errno;\n\tif (\"code\" in err && typeof err.code === \"string\") return err.code;\n\treturn undefined;\n}\n\nfunction titlePattern(query: string): string {\n\treturn `%${query.replace(/[\\\\%_]/g, \"\\\\$&\")}%`;\n}\n\nconst USER_COLUMNS = `\n\tid,\n\tlogin,\n\tavatar_url AS \"avatarUrl\",\n\tcreated_at AS \"createdAt\",\n\tupdated_at AS \"updatedAt\"\n`;\n\nconst SESSION_COLUMNS = `\n\tid,\n\tuser_id AS \"userId\",\n\texpires_at AS \"expiresAt\",\n\tcreated_at AS \"createdAt\"\n`;\n\nconst CHANNEL_COLUMNS = `\n\tchannels.id,\n\tchannels.repository_id AS \"repositoryId\",\n\tchannels.repository_owner AS \"repositoryOwner\",\n\tchannels.repository_name AS \"repositoryName\",\n\tchannels.title,\n\t(\n\t\tSELECT channel_slugs.slug\n\t\tFROM channel_slugs\n\t\tWHERE channel_slugs.channel_id = channels.id AND channel_slugs.canonical\n\t) AS slug,\n\tchannels.created_by AS \"createdBy\",\n\tchannels.revision,\n\tchannels.created_at AS \"createdAt\",\n\tchannels.updated_at AS \"updatedAt\"\n`;\n\nconst CHANNEL_RETURNING = `\n\tid,\n\trepository_id AS \"repositoryId\",\n\trepository_owner AS \"repositoryOwner\",\n\trepository_name AS \"repositoryName\",\n\ttitle,\n\tcreated_by AS \"createdBy\",\n\trevision,\n\tcreated_at AS \"createdAt\",\n\tupdated_at AS \"updatedAt\"\n`;\n\nconst SNAPSHOT_COLUMNS = `\n\tchannel_id AS \"channelId\",\n\tgeneration,\n\trevision,\n\tthrough_sequence AS \"throughSequence\",\n\tepoch,\n\tsource,\n\tsource_hash AS \"sourceHash\",\n\tdocument,\n\tsidecar,\n\tcreated_at AS \"createdAt\"\n`;\n\nconst UPDATE_COLUMNS = `\n\tchannel_id AS \"channelId\",\n\tsequence,\n\trevision,\n\tepoch,\n\tupdate\n`;\n\nconst EVENT_COLUMNS = `\n\tid,\n\tchannel_id AS \"channelId\",\n\tsequence,\n\tordinal,\n\tkind,\n\tpayload,\n\tcreated_at AS \"createdAt\"\n`;\n\nconst AGENT_COLUMNS = `\n\tchannel_id AS \"channelId\",\n\towner_session_id AS \"ownerSessionId\",\n\tgeneration,\n\tsummary,\n\ttranscript_cursor AS \"transcriptCursor\",\n\tstatus,\n\tupdated_at AS \"updatedAt\"\n`;\n\nconst LEASE_COLUMNS = `\n\tname,\n\towner,\n\tfencing,\n\texpires_at AS \"expiresAt\"\n`;\n\n/** PostgreSQL is the reference implementation, not part of the storage contract. */\nexport class PostgresStorage implements StorageAdapter {\n\treadonly driver = \"postgres\";\n\treadonly #sql: SQL;\n\n\tconstructor(url: string) {\n\t\tthis.#sql = new SQL(url, { connectionTimeout: 10, idleTimeout: 30, max: 10 });\n\t\tthis.navigation = new PostgresNavigationStore(\n\t\t\tthis.#sql,\n\t\t\t(action, execute) => this.#run(action, execute),\n\t\t);\n\t\tthis.jobs = new PostgresBackgroundJobStore(\n\t\t\tthis.#sql,\n\t\t\t(action, execute) => this.#run(action, execute),\n\t\t\tasync (transaction, lease) => {\n\t\t\t\tawait this.#assertLease(transaction, lease);\n\t\t\t\t// The deployment owns one writer; preserve that ordering even if a caller\n\t\t\t\t// supplies another valid lease name to the provider-neutral port.\n\t\t\t\tawait transaction`SELECT pg_advisory_xact_lock(2043237432)`;\n\t\t\t\tawait this.#assertLease(transaction, lease);\n\t\t\t},\n\t\t);\n\t}\n\n\treadonly users: UserStore = {\n\t\tput: input =>\n\t\t\tthis.#run(\"save user\", async () => {\n\t\t\t\tlet [saved] = await this.#sql`\n\t\t\t\tINSERT INTO users (id, login, avatar_url, created_at, updated_at)\n\t\t\t\tVALUES (${input.id}, ${input.login}, ${input.avatarUrl}, ${input.now}, ${input.now})\n\t\t\t\tON CONFLICT (id) DO UPDATE SET\n\t\t\t\t\tlogin = EXCLUDED.login,\n\t\t\t\t\tavatar_url = EXCLUDED.avatar_url,\n\t\t\t\t\tupdated_at = EXCLUDED.updated_at\n\t\t\t\tRETURNING ${this.#sql.unsafe(USER_COLUMNS)}\n\t\t\t`;\n\t\t\t\tif (!saved) throw corrupt(\"saving a user returned no record\");\n\t\t\t\treturn user(saved);\n\t\t\t}),\n\t\tget: id =>\n\t\t\tthis.#run(\"read user\", async () => {\n\t\t\t\tlet [found] = await this.#sql`\n\t\t\t\tSELECT ${this.#sql.unsafe(USER_COLUMNS)} FROM users WHERE id = ${id}\n\t\t\t`;\n\t\t\t\treturn found ? user(found) : undefined;\n\t\t\t}),\n\t};\n\n\treadonly sessions: SessionStore = {\n\t\tcreate: input =>\n\t\t\tthis.#run(\"create login session\", async () => {\n\t\t\t\tawait this.#sql`\n\t\t\t\tINSERT INTO web_sessions (\n\t\t\t\t\tid, user_id, expires_at, created_at\n\t\t\t\t) VALUES (\n\t\t\t\t\t${input.id}, ${input.userId}, ${input.expiresAt}, ${input.createdAt}\n\t\t\t\t)\n\t\t\t`;\n\t\t\t}),\n\t\tget: (id, now) =>\n\t\t\tthis.#run(\"read login session\", async () => {\n\t\t\t\tlet [found] = await this.#sql`\n\t\t\t\tSELECT ${this.#sql.unsafe(SESSION_COLUMNS)}\n\t\t\t\tFROM web_sessions\n\t\t\t\tWHERE id = ${id} AND expires_at > ${now}\n\t\t\t`;\n\t\t\t\treturn found ? session(found) : undefined;\n\t\t\t}),\n\t\tdelete: id =>\n\t\t\tthis.#run(\"delete login session\", () =>\n\t\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\t\tawait transaction`\n\t\t\t\t\tUPDATE agent_state\n\t\t\t\t\tSET owner_session_id = NULL, status = 'unavailable', updated_at = now()\n\t\t\t\t\tWHERE owner_session_id = ${id}\n\t\t\t\t`;\n\t\t\t\t\tlet deleted = await transaction<{ id: string }[]>`\n\t\t\t\t\tDELETE FROM web_sessions WHERE id = ${id} RETURNING id\n\t\t\t\t`;\n\t\t\t\t\treturn deleted.length > 0;\n\t\t\t\t})),\n\t\tdeleteExpired: now =>\n\t\t\tthis.#run(\"delete expired login sessions\", () =>\n\t\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\t\tawait transaction`\n\t\t\t\t\tUPDATE agent_state\n\t\t\t\t\tSET owner_session_id = NULL, status = 'unavailable', updated_at = ${now}\n\t\t\t\t\tWHERE owner_session_id IN (\n\t\t\t\t\t\tSELECT id FROM web_sessions WHERE expires_at <= ${now}\n\t\t\t\t\t)\n\t\t\t\t`;\n\t\t\t\t\tlet deleted = await transaction<{ id: string }[]>`\n\t\t\t\t\tDELETE FROM web_sessions WHERE expires_at <= ${now} RETURNING id\n\t\t\t\t`;\n\t\t\t\t\treturn deleted.length;\n\t\t\t\t})),\n\t\tdeleteAll: (now, held, ttlMs) =>\n\t\t\tthis.#run(\"delete all login sessions\", () =>\n\t\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\t\tawait this.#assertLease(transaction, held);\n\t\t\t\t\tawait transaction`\n\t\t\t\t\tUPDATE agent_state\n\t\t\t\t\tSET owner_session_id = NULL, status = 'unavailable', updated_at = ${now}\n\t\t\t\t\tWHERE owner_session_id IS NOT NULL\n\t\t\t\t`;\n\t\t\t\t\tlet deleted = await transaction<{ id: string }[]>`\n\t\t\t\t\tDELETE FROM web_sessions RETURNING id\n\t\t\t\t\t`;\n\t\t\t\t\tlet [renewed] = await transaction`\n\t\t\t\t\tUPDATE storage_leases\n\t\t\t\t\tSET expires_at = clock_timestamp() + (${ttlMs} * interval '1 millisecond')\n\t\t\t\t\tWHERE name = ${held.name}\n\t\t\t\t\t\tAND owner = ${held.owner}\n\t\t\t\t\t\tAND fencing = ${held.fencing}\n\t\t\t\t\tRETURNING ${transaction.unsafe(LEASE_COLUMNS)}\n\t\t\t\t\t`;\n\t\t\t\t\tif (!renewed) throw conflict(`storage lease ${held.name} is no longer held`);\n\t\t\t\t\treturn { deleted: deleted.length, lease: lease(renewed) };\n\t\t\t\t})),\n\t};\n\n\treadonly navigation: NavigationStore;\n\treadonly jobs: BackgroundJobStore;\n\n\treadonly channels: ChannelStore = {\n\t\tcreate: input => this.#createChannel(input),\n\t\tget: id =>\n\t\t\tthis.#run(\"read channel\", async () => {\n\t\t\t\tlet [found] = await this.#sql`\n\t\t\t\tSELECT ${this.#sql.unsafe(CHANNEL_COLUMNS)} FROM channels WHERE id = ${id}\n\t\t\t`;\n\t\t\t\treturn found ? channel(found) : undefined;\n\t\t\t}),\n\t\tresolve: (repositoryId, slug) =>\n\t\t\tthis.#run(\"resolve channel slug\", async () => {\n\t\t\t\tlet [found] = await this.#sql`\n\t\t\t\t\tSELECT ${this.#sql.unsafe(CHANNEL_COLUMNS)}\n\t\t\t\t\tFROM channel_slugs\n\t\t\t\t\tJOIN channels\n\t\t\t\t\t\tON channels.id = channel_slugs.channel_id\n\t\t\t\t\t\tAND channels.repository_id = channel_slugs.repository_id\n\t\t\t\t\tWHERE channel_slugs.repository_id = ${repositoryId}\n\t\t\t\t\t\tAND channel_slugs.slug = ${slug}\n\t\t\t\t`;\n\t\t\t\treturn found ? channel(found) : undefined;\n\t\t\t}),\n\t\trename: input => this.#renameChannel(input),\n\t\tlist: (repositoryId, limit, after, query) =>\n\t\t\tthis.#listChannels(repositoryId, limit, after, query),\n\t\tscan: (repositoryId, limit, after) => this.#scanChannels(repositoryId, limit, after),\n\t\tclaimAgentOwner: (channelId, sessionId, now) =>\n\t\t\tthis.#claimAgentOwner(channelId, sessionId, now),\n\t\tclearAgentOwner: (channelId, expectedSessionId, expectedGeneration, now) =>\n\t\t\tthis.#clearAgentOwner(channelId, expectedSessionId, expectedGeneration, now),\n\t\tupdateAgentContext: input => this.#updateAgentContext(input),\n\t};\n\n\treadonly collaboration: CollaborationStore = {\n\t\tload: (channelId, now) => this.#load(channelId, now),\n\t\tcommit: input => this.#commit(input),\n\t\treplace: input => this.#replace(input),\n\t\tcheckpoint: input => this.#checkpoint(input),\n\t};\n\n\treadonly leases: LeaseStore = {\n\t\tacquire: (name, owner, ttlMs) => this.#acquire(name, owner, ttlMs),\n\t\trenew: (held, ttlMs) => this.#renew(held, ttlMs),\n\t\trelease: held => this.#release(held),\n\t};\n\n\tasync migrate(): Promise {\n\t\tawait this.#run(\"migrate storage\", async () => {\n\t\t\tawait this.#sql.connect();\n\t\t\tawait migrate(this.#sql);\n\t\t});\n\t}\n\n\tasync health(): Promise {\n\t\tawait this.#run(\"check storage\", async () => {\n\t\t\tawait this.#sql`SELECT 1`;\n\t\t\tawait verifyMigrations(this.#sql);\n\t\t});\n\t}\n\n\tasync close(): Promise {\n\t\tawait this.#sql.close({ timeout: 5 });\n\t}\n\n\tasync #run(action: string, execute: () => Promise): Promise {\n\t\ttry {\n\t\t\treturn await execute();\n\t\t} catch (err) {\n\t\t\tif (err instanceof StorageError) throw err;\n\t\t\tlet code = postgresCode(err);\n\t\t\tif (code === \"23505\") throw conflict(`${action} conflicts with an existing record`);\n\t\t\tif (code === \"23503\") throw missing(`${action} refers to a missing record`);\n\t\t\tthrow unavailable(`cannot ${action}`, err);\n\t\t}\n\t}\n\n\t#createChannel(input: CreateChannel): Promise {\n\t\treturn this.#run(\"create channel\", () =>\n\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\tlet [saved] = await transaction`\n\t\t\t\tINSERT INTO channels (\n\t\t\t\t\tid, repository_id, repository_owner, repository_name, title,\n\t\t\t\t\tcreated_by, revision, next_sequence, created_at, updated_at\n\t\t\t\t) VALUES (\n\t\t\t\t\t${input.id}, ${input.repositoryId}, ${input.repositoryOwner},\n\t\t\t\t\t${input.repositoryName}, ${input.title}, ${input.createdBy}, 0, 1,\n\t\t\t\t\t${input.now}, ${input.now}\n\t\t\t\t)\n\t\t\t\tRETURNING ${transaction.unsafe(CHANNEL_RETURNING)}\n\t\t\t`;\n\t\t\t\tif (!saved) throw corrupt(\"creating a channel returned no record\");\n\t\t\t\tlet slug = await this.#reserveSlug(\n\t\t\t\t\ttransaction,\n\t\t\t\t\tinput.repositoryId,\n\t\t\t\t\tinput.id,\n\t\t\t\t\tinput.title,\n\t\t\t\t\tinput.now,\n\t\t\t\t);\n\t\t\t\tawait transaction`\n\t\t\t\tINSERT INTO channel_state (channel_id, sidecar)\n\t\t\t\tVALUES (\n\t\t\t\t\t${input.id},\n\t\t\t\t\t${input.initial ? JSON.stringify(input.initial.sidecar) : \"null\"}::jsonb\n\t\t\t\t)\n\t\t\t`;\n\t\t\t\tif (input.initial) {\n\t\t\t\t\tawait transaction`\n\t\t\t\t\t\tINSERT INTO channel_snapshots (\n\t\t\t\t\t\t\tchannel_id, generation, revision, through_sequence, epoch, source,\n\t\t\t\t\t\t\tsource_hash, document, sidecar, created_at\n\t\t\t\t\t\t) VALUES (\n\t\t\t\t\t\t\t${input.id}, ${input.initial.generation}, 0, 0, ${input.initial.epoch},\n\t\t\t\t\t\t\t${input.initial.source}, ${input.initial.sourceHash},\n\t\t\t\t\t\t\t${input.initial.document}, ${JSON.stringify(input.initial.sidecar)}::jsonb,\n\t\t\t\t\t\t\t${input.now}\n\t\t\t\t\t\t)\n\t\t\t\t\t`;\n\t\t\t\t}\n\t\t\t\treturn channel({ ...saved, slug });\n\t\t\t}));\n\t}\n\n\t#renameChannel(input: RenameChannel): Promise {\n\t\treturn this.#run(\"rename channel\", () =>\n\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\tlet [current] = await transaction`\n\t\t\t\t\tSELECT ${transaction.unsafe(CHANNEL_COLUMNS)}\n\t\t\t\t\tFROM channels\n\t\t\t\t\tWHERE id = ${input.id}\n\t\t\t\t\tFOR UPDATE\n\t\t\t\t`;\n\t\t\t\tif (!current) throw missing(`channel ${input.id} does not exist`);\n\t\t\t\tlet existing = channel(current);\n\t\t\t\tif (existing.title === input.title) {\n\t\t\t\t\treturn { channel: existing, changed: false };\n\t\t\t\t}\n\t\t\t\tlet updatedAt = input.now > existing.updatedAt\n\t\t\t\t\t? input.now\n\t\t\t\t\t: new Date(existing.updatedAt.getTime() + 1);\n\t\t\t\tlet [saved] = await transaction`\n\t\t\t\t\tUPDATE channels\n\t\t\t\t\tSET title = ${input.title}, updated_at = ${updatedAt}\n\t\t\t\t\tWHERE id = ${input.id}\n\t\t\t\t\tRETURNING ${transaction.unsafe(CHANNEL_RETURNING)}\n\t\t\t\t`;\n\t\t\t\tif (!saved) throw corrupt(\"renaming a channel returned no record\");\n\t\t\t\tlet slug = await this.#reserveSlug(\n\t\t\t\t\ttransaction,\n\t\t\t\t\texisting.repositoryId,\n\t\t\t\t\texisting.id,\n\t\t\t\t\tinput.title,\n\t\t\t\t\tinput.now,\n\t\t\t\t);\n\t\t\t\treturn { channel: channel({ ...saved, slug }), changed: true };\n\t\t\t}));\n\t}\n\n\tasync #reserveSlug(\n\t\ttransaction: TransactionSQL,\n\t\trepositoryId: string,\n\t\tchannelId: string,\n\t\ttitle: string,\n\t\tnow: Date,\n\t): Promise {\n\t\tlet base = documentSlug(title);\n\t\tfor (let index = 1;; index++) {\n\t\t\tlet candidate = documentSlugCandidate(base, index);\n\t\t\tlet [inserted] = await transaction<{ channelId: string }[]>`\n\t\t\t\tINSERT INTO channel_slugs (\n\t\t\t\t\trepository_id, slug, channel_id, canonical, created_at\n\t\t\t\t) VALUES (${repositoryId}, ${candidate}, ${channelId}, false, ${now})\n\t\t\t\tON CONFLICT DO NOTHING\n\t\t\t\tRETURNING channel_id AS \"channelId\"\n\t\t\t`;\n\t\t\tif (!inserted) {\n\t\t\t\tlet [existing] = await transaction<{ channelId: string }[]>`\n\t\t\t\t\tSELECT channel_id AS \"channelId\"\n\t\t\t\t\tFROM channel_slugs\n\t\t\t\t\tWHERE repository_id = ${repositoryId} AND slug = ${candidate}\n\t\t\t\t`;\n\t\t\t\tif (existing?.channelId !== channelId) continue;\n\t\t\t}\n\n\t\t\tawait transaction`\n\t\t\t\tUPDATE channel_slugs SET canonical = false\n\t\t\t\tWHERE channel_id = ${channelId} AND canonical\n\t\t\t`;\n\t\t\tlet [promoted] = await transaction<{ slug: string }[]>`\n\t\t\t\tUPDATE channel_slugs SET canonical = true\n\t\t\t\tWHERE repository_id = ${repositoryId}\n\t\t\t\t\tAND slug = ${candidate}\n\t\t\t\t\tAND channel_id = ${channelId}\n\t\t\t\tRETURNING slug\n\t\t\t`;\n\t\t\tif (!promoted) throw corrupt(`could not reserve a slug for channel ${channelId}`);\n\t\t\treturn promoted.slug;\n\t\t}\n\t}\n\n\t#listChannels(\n\t\trepositoryId: string,\n\t\tlimit: number,\n\t\tafter?: ChannelCursor,\n\t\tquery?: string,\n\t): Promise {\n\t\treturn this.#run(\"list channels\", async () => {\n\t\t\tlet count = Math.min(100, Math.max(1, limit));\n\t\t\tlet pattern = query ? titlePattern(query) : \"\";\n\t\t\tlet rows = after\n\t\t\t\t? await this.#sql`\n\t\t\t\t\tSELECT ${this.#sql.unsafe(CHANNEL_COLUMNS)}\n\t\t\t\t\tFROM channels\n\t\t\t\t\tWHERE repository_id = ${repositoryId}\n\t\t\t\t\t\tAND (${!query} OR title ILIKE ${pattern} ESCAPE '\\\\')\n\t\t\t\t\t\tAND (\n\t\t\t\t\t\t\tupdated_at < ${after.updatedAt}\n\t\t\t\t\t\t\tOR (updated_at = ${after.updatedAt} AND id > ${after.id})\n\t\t\t\t\t\t)\n\t\t\t\t\tORDER BY updated_at DESC, id ASC\n\t\t\t\t\tLIMIT ${count + 1}\n\t\t\t\t`\n\t\t\t\t: await this.#sql`\n\t\t\t\t\tSELECT ${this.#sql.unsafe(CHANNEL_COLUMNS)}\n\t\t\t\t\tFROM channels\n\t\t\t\t\tWHERE repository_id = ${repositoryId}\n\t\t\t\t\t\tAND (${!query} OR title ILIKE ${pattern} ESCAPE '\\\\')\n\t\t\t\t\tORDER BY updated_at DESC, id ASC\n\t\t\t\t\tLIMIT ${count + 1}\n\t\t\t\t`;\n\t\t\tlet more = rows.length > count;\n\t\t\tlet page = rows.slice(0, count).map(channel);\n\t\t\tlet last = page.at(-1);\n\t\t\treturn {\n\t\t\t\tchannels: page,\n\t\t\t\tnext: more && last ? { updatedAt: last.updatedAt, id: last.id } : undefined,\n\t\t\t};\n\t\t});\n\t}\n\n\t#scanChannels(\n\t\trepositoryId: string,\n\t\tlimit: number,\n\t\tafter?: ChannelScanCursor,\n\t): Promise {\n\t\treturn this.#run(\"scan channels\", async () => {\n\t\t\tlet count = Math.min(100, Math.max(1, limit));\n\t\t\tlet rows = after\n\t\t\t\t? await this.#sql`\n\t\t\t\t\tSELECT ${this.#sql.unsafe(CHANNEL_COLUMNS)}\n\t\t\t\t\tFROM channels\n\t\t\t\t\tWHERE repository_id = ${repositoryId}\n\t\t\t\t\t\tAND (\n\t\t\t\t\t\t\tcreated_at < ${after.createdAt}\n\t\t\t\t\t\t\tOR (created_at = ${after.createdAt} AND id > ${after.id})\n\t\t\t\t\t\t)\n\t\t\t\t\tORDER BY created_at DESC, id ASC\n\t\t\t\t\tLIMIT ${count + 1}\n\t\t\t\t`\n\t\t\t\t: await this.#sql`\n\t\t\t\t\tSELECT ${this.#sql.unsafe(CHANNEL_COLUMNS)}\n\t\t\t\t\tFROM channels\n\t\t\t\t\tWHERE repository_id = ${repositoryId}\n\t\t\t\t\tORDER BY created_at DESC, id ASC\n\t\t\t\t\tLIMIT ${count + 1}\n\t\t\t\t`;\n\t\t\tlet more = rows.length > count;\n\t\t\tlet page = rows.slice(0, count).map(channel);\n\t\t\tlet last = page.at(-1);\n\t\t\treturn {\n\t\t\t\tchannels: page,\n\t\t\t\tnext: more && last ? { createdAt: last.createdAt, id: last.id } : undefined,\n\t\t\t};\n\t\t});\n\t}\n\n\t#claimAgentOwner(channelId: string, sessionId: string, now: Date): Promise {\n\t\treturn this.#run(\"claim agent owner\", () =>\n\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\tlet [active] = await transaction<{ id: string }[]>`\n\t\t\t\tSELECT id FROM web_sessions WHERE id = ${sessionId} AND expires_at > ${now}\n\t\t\t`;\n\t\t\t\tif (!active) throw missing(`session ${sessionId} is not active`);\n\n\t\t\t\tawait transaction`\n\t\t\t\tUPDATE agent_state\n\t\t\t\tSET owner_session_id = NULL, status = 'unavailable', updated_at = ${now}\n\t\t\t\tWHERE channel_id = ${channelId}\n\t\t\t\t\tAND owner_session_id IN (\n\t\t\t\t\t\tSELECT id FROM web_sessions WHERE expires_at <= ${now}\n\t\t\t\t\t)\n\t\t\t`;\n\t\t\t\tlet [claimed] = await transaction`\n\t\t\t\tINSERT INTO agent_state (\n\t\t\t\t\tchannel_id, owner_session_id, generation, summary,\n\t\t\t\t\ttranscript_cursor, status, updated_at\n\t\t\t\t) VALUES (${channelId}, ${sessionId}, 1, '', 0, 'ready', ${now})\n\t\t\t\tON CONFLICT (channel_id) DO UPDATE SET\n\t\t\t\t\towner_session_id = EXCLUDED.owner_session_id,\n\t\t\t\t\tgeneration = agent_state.generation + 1,\n\t\t\t\t\tstatus = 'ready',\n\t\t\t\t\tupdated_at = EXCLUDED.updated_at\n\t\t\t\tWHERE agent_state.owner_session_id IS NULL\n\t\t\t\tRETURNING ${transaction.unsafe(AGENT_COLUMNS)}\n\t\t\t`;\n\t\t\t\tif (claimed) return agent(claimed);\n\t\t\t\tlet [existing] = await transaction`\n\t\t\t\tSELECT ${transaction.unsafe(AGENT_COLUMNS)}\n\t\t\t\tFROM agent_state WHERE channel_id = ${channelId}\n\t\t\t`;\n\t\t\t\tif (!existing) throw corrupt(\"claiming an agent owner returned no record\");\n\t\t\t\treturn agent(existing);\n\t\t\t}));\n\t}\n\n\t#clearAgentOwner(\n\t\tchannelId: string,\n\t\texpectedSessionId: string,\n\t\texpectedGeneration: number,\n\t\tnow: Date,\n\t): Promise {\n\t\treturn this.#run(\"clear agent owner\", async () => {\n\t\t\tlet changed = await this.#sql<{ channelId: string }[]>`\n\t\t\t\tUPDATE agent_state\n\t\t\t\tSET owner_session_id = NULL, status = 'unavailable', updated_at = ${now}\n\t\t\t\tWHERE channel_id = ${channelId}\n\t\t\t\t\tAND owner_session_id = ${expectedSessionId}\n\t\t\t\t\tAND generation = ${expectedGeneration}\n\t\t\t\tRETURNING channel_id AS \"channelId\"\n\t\t\t`;\n\t\t\treturn changed.length > 0;\n\t\t});\n\t}\n\n\t#updateAgentContext(input: UpdateAgentContext): Promise {\n\t\treturn this.#run(\"update agent context\", async () => {\n\t\t\tlet [saved] = await this.#sql`\n\t\t\t\tUPDATE agent_state\n\t\t\t\tSET summary = ${input.summary},\n\t\t\t\t\ttranscript_cursor = ${input.transcriptCursor},\n\t\t\t\t\tstatus = ${input.status},\n\t\t\t\t\tupdated_at = ${input.now}\n\t\t\t\tWHERE channel_id = ${input.channelId}\n\t\t\t\t\tAND owner_session_id = ${input.ownerSessionId}\n\t\t\t\t\tAND generation = ${input.generation}\n\t\t\t\t\tAND EXISTS (\n\t\t\t\t\t\tSELECT 1 FROM web_sessions\n\t\t\t\t\t\tWHERE id = ${input.ownerSessionId} AND expires_at > ${input.now}\n\t\t\t\t\t)\n\t\t\t\tRETURNING ${this.#sql.unsafe(AGENT_COLUMNS)}\n\t\t\t`;\n\t\t\tif (!saved) throw conflict(`agent owner changed for channel ${input.channelId}`);\n\t\t\treturn agent(saved);\n\t\t});\n\t}\n\n\t#load(channelId: string, now: Date): Promise {\n\t\treturn this.#run(\n\t\t\t\"load channel\",\n\t\t\t() =>\n\t\t\t\tthis.#sql.begin(\"isolation level repeatable read read only\", async transaction => {\n\t\t\t\t\tlet [channelRow] = await transaction`\n\t\t\t\tSELECT ${transaction.unsafe(CHANNEL_COLUMNS)}, next_sequence AS \"nextSequence\"\n\t\t\t\tFROM channels WHERE id = ${channelId}\n\t\t\t`;\n\t\t\t\t\tif (!channelRow) return undefined;\n\t\t\t\t\tlet [snapshotRow] = await transaction`\n\t\t\t\tSELECT ${transaction.unsafe(SNAPSHOT_COLUMNS)}\n\t\t\t\tFROM channel_snapshots WHERE channel_id = ${channelId}\n\t\t\t`;\n\t\t\t\t\tlet through = snapshotRow\n\t\t\t\t\t\t? integer(snapshotRow.throughSequence, \"snapshot sequence\")\n\t\t\t\t\t\t: 0;\n\t\t\t\t\tlet updateRows = await transaction`\n\t\t\t\tSELECT ${transaction.unsafe(UPDATE_COLUMNS)}\n\t\t\t\tFROM channel_updates\n\t\t\t\tWHERE channel_id = ${channelId} AND sequence > ${through}\n\t\t\t\tORDER BY sequence ASC\n\t\t\t`;\n\t\t\t\t\tlet eventRows = await transaction`\n\t\t\t\tSELECT ${transaction.unsafe(EVENT_COLUMNS)}\n\t\t\t\tFROM channel_events\n\t\t\t\tWHERE channel_id = ${channelId}\n\t\t\t\tORDER BY sequence ASC, ordinal ASC\n\t\t\t`;\n\t\t\t\t\tlet [state] = await transaction<{ sidecar: unknown }[]>`\n\t\t\t\tSELECT sidecar FROM channel_state WHERE channel_id = ${channelId}\n\t\t\t`;\n\t\t\t\t\tif (!state) throw corrupt(`channel ${channelId} has no sidecar state`);\n\t\t\t\t\tlet [agentRow] = await transaction`\n\t\t\t\tSELECT ${transaction.unsafe(AGENT_COLUMNS)}\n\t\t\t\tFROM agent_state WHERE channel_id = ${channelId}\n\t\t\t`;\n\t\t\t\t\tlet storedAgent = agentRow ? agent(agentRow) : undefined;\n\t\t\t\t\tif (storedAgent?.ownerSessionId) {\n\t\t\t\t\t\tlet [active] = await transaction<{ id: string }[]>`\n\t\t\t\t\tSELECT id FROM web_sessions\n\t\t\t\t\tWHERE id = ${storedAgent.ownerSessionId} AND expires_at > ${now}\n\t\t\t\t`;\n\t\t\t\t\t\tif (!active) {\n\t\t\t\t\t\t\tstoredAgent = {\n\t\t\t\t\t\t\t\t...storedAgent,\n\t\t\t\t\t\t\t\townerSessionId: undefined,\n\t\t\t\t\t\t\t\tstatus: \"unavailable\",\n\t\t\t\t\t\t\t};\n\t\t\t\t\t\t}\n\t\t\t\t\t}\n\t\t\t\t\tlet nextSequence = integer(channelRow.nextSequence ?? 1, \"channel sequence\");\n\t\t\t\t\tif (nextSequence < 1) throw corrupt(\"storage returned an invalid channel sequence\");\n\t\t\t\t\treturn {\n\t\t\t\t\t\tchannel: channel(channelRow),\n\t\t\t\t\t\tlatestSequence: nextSequence - 1,\n\t\t\t\t\t\tsnapshot: snapshotRow ? snapshot(snapshotRow) : undefined,\n\t\t\t\t\t\tupdates: updateRows.map(update),\n\t\t\t\t\t\tevents: eventRows.map(event),\n\t\t\t\t\t\tsidecar: json(state.sidecar, \"channel sidecar\"),\n\t\t\t\t\t\tagent: storedAgent,\n\t\t\t\t\t};\n\t\t\t\t}),\n\t\t);\n\t}\n\n\t#commit(input: CommitChannel): Promise {\n\t\treturn this.#run(\"commit channel\", () =>\n\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\tawait this.#assertLease(transaction, input.lease);\n\t\t\t\tlet [locked] = await transaction`\n\t\t\t\tSELECT revision, next_sequence AS \"nextSequence\"\n\t\t\t\tFROM channels\n\t\t\t\tWHERE id = ${input.channelId}\n\t\t\t\tFOR UPDATE\n\t\t\t`;\n\t\t\t\tif (!locked) throw missing(`channel ${input.channelId} does not exist`);\n\t\t\t\tlet [previous] = await transaction`\n\t\t\t\tSELECT revision, sequence\n\t\t\t\tFROM channel_operations\n\t\t\t\tWHERE channel_id = ${input.channelId} AND operation_id = ${input.operationId}\n\t\t\t`;\n\t\t\t\tif (previous) {\n\t\t\t\t\treturn {\n\t\t\t\t\t\trevision: integer(previous.revision, \"operation revision\"),\n\t\t\t\t\t\tsequence: integer(previous.sequence, \"operation sequence\"),\n\t\t\t\t\t\trepeated: true,\n\t\t\t\t\t};\n\t\t\t\t}\n\t\t\t\tlet current = integer(locked.revision, \"channel revision\");\n\t\t\t\tif (current !== input.expectedRevision) {\n\t\t\t\t\tthrow conflict(\n\t\t\t\t\t\t`channel ${input.channelId} is at revision ${current}, expected ${input.expectedRevision}`,\n\t\t\t\t\t);\n\t\t\t\t}\n\t\t\t\tlet sequence = integer(locked.nextSequence, \"channel sequence\");\n\t\t\t\tlet revision = current + 1;\n\t\t\t\tawait transaction`\n\t\t\t\tINSERT INTO channel_operations (channel_id, operation_id, sequence, revision)\n\t\t\t\tVALUES (${input.channelId}, ${input.operationId}, ${sequence}, ${revision})\n\t\t\t`;\n\t\t\t\tif (input.update) {\n\t\t\t\t\tawait transaction`\n\t\t\t\t\tINSERT INTO channel_updates (channel_id, sequence, revision, epoch, update)\n\t\t\t\t\tVALUES (\n\t\t\t\t\t\t${input.channelId}, ${sequence}, ${revision}, ${input.epoch}, ${input.update}\n\t\t\t\t\t)\n\t\t\t\t`;\n\t\t\t\t}\n\t\t\t\tif (input.sidecar !== undefined) {\n\t\t\t\t\tawait transaction`\n\t\t\t\t\tUPDATE channel_state\n\t\t\t\t\tSET sidecar = ${JSON.stringify(input.sidecar)}::jsonb\n\t\t\t\t\tWHERE channel_id = ${input.channelId}\n\t\t\t\t`;\n\t\t\t\t}\n\t\t\t\tfor (let [ordinal, item] of input.events.entries()) {\n\t\t\t\t\tawait transaction`\n\t\t\t\t\tINSERT INTO channel_events (\n\t\t\t\t\t\tchannel_id, sequence, ordinal, id, kind, payload, created_at\n\t\t\t\t\t) VALUES (\n\t\t\t\t\t\t${input.channelId}, ${sequence}, ${ordinal}, ${item.id}, ${item.kind},\n\t\t\t\t\t\t${JSON.stringify(item.payload)}::jsonb, ${item.createdAt}\n\t\t\t\t\t)\n\t\t\t\t`;\n\t\t\t\t}\n\t\t\t\tawait transaction`\n\t\t\t\tUPDATE channels\n\t\t\t\tSET revision = ${revision}, next_sequence = ${sequence + 1},\n\t\t\t\t\tupdated_at = GREATEST(updated_at, ${input.now})\n\t\t\t\tWHERE id = ${input.channelId}\n\t\t\t`;\n\t\t\t\treturn { revision, sequence, repeated: false };\n\t\t\t}));\n\t}\n\n\t#checkpoint(input: SaveCheckpoint): Promise {\n\t\treturn this.#run(\"checkpoint channel\", () =>\n\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\tawait this.#assertLease(transaction, input.lease);\n\t\t\t\tlet [locked] = await transaction`\n\t\t\t\tSELECT revision, next_sequence AS \"nextSequence\"\n\t\t\t\tFROM channels\n\t\t\t\tWHERE id = ${input.channelId}\n\t\t\t\tFOR UPDATE\n\t\t\t`;\n\t\t\t\tif (!locked) throw missing(`channel ${input.channelId} does not exist`);\n\t\t\t\tlet revision = integer(locked.revision, \"channel revision\");\n\t\t\t\tlet lastSequence = integer(locked.nextSequence, \"channel sequence\") - 1;\n\t\t\t\tif (revision !== input.expectedRevision || input.revision !== input.expectedRevision) {\n\t\t\t\t\tthrow conflict(`channel ${input.channelId} changed before its checkpoint`);\n\t\t\t\t}\n\t\t\t\tif (input.throughSequence > lastSequence) {\n\t\t\t\t\tthrow conflict(`checkpoint for channel ${input.channelId} is ahead of its journal`);\n\t\t\t\t}\n\t\t\t\tlet [previous] = await transaction<{ throughSequence: Integer }[]>`\n\t\t\t\tSELECT through_sequence AS \"throughSequence\"\n\t\t\t\tFROM channel_snapshots\n\t\t\t\tWHERE channel_id = ${input.channelId}\n\t\t\t`;\n\t\t\t\tif (\n\t\t\t\t\tprevious\n\t\t\t\t\t&& integer(previous.throughSequence, \"snapshot sequence\") > input.throughSequence\n\t\t\t\t) throw conflict(`channel ${input.channelId} already has a newer checkpoint`);\n\n\t\t\t\tawait transaction`\n\t\t\t\tINSERT INTO channel_snapshots (\n\t\t\t\t\tchannel_id, generation, revision, through_sequence, epoch, source,\n\t\t\t\t\tsource_hash, document, sidecar, created_at\n\t\t\t\t) VALUES (\n\t\t\t\t\t${input.channelId}, ${input.generation}, ${input.revision},\n\t\t\t\t\t${input.throughSequence}, ${input.epoch}, ${input.source}, ${input.sourceHash},\n\t\t\t\t\t${input.document}, ${JSON.stringify(input.sidecar)}::jsonb, ${input.createdAt}\n\t\t\t\t)\n\t\t\t\tON CONFLICT (channel_id) DO UPDATE SET\n\t\t\t\t\tgeneration = EXCLUDED.generation,\n\t\t\t\t\trevision = EXCLUDED.revision,\n\t\t\t\t\tthrough_sequence = EXCLUDED.through_sequence,\n\t\t\t\t\tepoch = EXCLUDED.epoch,\n\t\t\t\t\tsource = EXCLUDED.source,\n\t\t\t\t\tsource_hash = EXCLUDED.source_hash,\n\t\t\t\t\tdocument = EXCLUDED.document,\n\t\t\t\t\tsidecar = EXCLUDED.sidecar,\n\t\t\t\t\tcreated_at = EXCLUDED.created_at\n\t\t\t`;\n\t\t\t\tawait transaction`\n\t\t\t\tDELETE FROM channel_updates\n\t\t\t\tWHERE channel_id = ${input.channelId} AND sequence <= ${input.throughSequence}\n\t\t\t`;\n\t\t\t}));\n\t}\n\n\t#replace(input: ReplaceChannel): Promise {\n\t\treturn this.#run(\"replace channel\", () =>\n\t\t\tthis.#sql.begin(async transaction => {\n\t\t\t\tawait this.#assertLease(transaction, input.lease);\n\t\t\t\tlet [locked] = await transaction`\n\t\t\t\t\tSELECT revision, next_sequence AS \"nextSequence\"\n\t\t\t\t\tFROM channels\n\t\t\t\t\tWHERE id = ${input.channelId}\n\t\t\t\t\tFOR UPDATE\n\t\t\t\t`;\n\t\t\t\tif (!locked) throw missing(`channel ${input.channelId} does not exist`);\n\t\t\t\tlet [previous] = await transaction`\n\t\t\t\t\tSELECT revision, sequence\n\t\t\t\t\tFROM channel_operations\n\t\t\t\t\tWHERE channel_id = ${input.channelId} AND operation_id = ${input.operationId}\n\t\t\t\t`;\n\t\t\t\tif (previous) {\n\t\t\t\t\treturn {\n\t\t\t\t\t\trevision: integer(previous.revision, \"operation revision\"),\n\t\t\t\t\t\tsequence: integer(previous.sequence, \"operation sequence\"),\n\t\t\t\t\t\trepeated: true,\n\t\t\t\t\t};\n\t\t\t\t}\n\t\t\t\tlet current = integer(locked.revision, \"channel revision\");\n\t\t\t\tif (current !== input.expectedRevision) {\n\t\t\t\t\tthrow conflict(\n\t\t\t\t\t\t`channel ${input.channelId} is at revision ${current}, expected ${input.expectedRevision}`,\n\t\t\t\t\t);\n\t\t\t\t}\n\t\t\t\tlet sequence = integer(locked.nextSequence, \"channel sequence\");\n\t\t\t\tlet revision = current + 1;\n\t\t\t\tawait transaction`\n\t\t\t\t\tINSERT INTO channel_operations (channel_id, operation_id, sequence, revision)\n\t\t\t\t\tVALUES (${input.channelId}, ${input.operationId}, ${sequence}, ${revision})\n\t\t\t\t`;\n\t\t\t\tawait transaction`\n\t\t\t\t\tUPDATE channel_state\n\t\t\t\t\tSET sidecar = ${JSON.stringify(input.sidecar)}::jsonb\n\t\t\t\t\tWHERE channel_id = ${input.channelId}\n\t\t\t\t`;\n\t\t\t\tawait transaction`\n\t\t\t\t\tINSERT INTO channel_snapshots (\n\t\t\t\t\t\tchannel_id, generation, revision, through_sequence, epoch, source,\n\t\t\t\t\t\tsource_hash, document, sidecar, created_at\n\t\t\t\t\t) VALUES (\n\t\t\t\t\t\t${input.channelId}, ${input.generation}, ${revision}, ${sequence},\n\t\t\t\t\t\t${input.epoch}, ${input.source}, ${input.sourceHash}, ${input.document},\n\t\t\t\t\t\t${JSON.stringify(input.sidecar)}::jsonb, ${input.now}\n\t\t\t\t\t)\n\t\t\t\t\tON CONFLICT (channel_id) DO UPDATE SET\n\t\t\t\t\t\tgeneration = EXCLUDED.generation,\n\t\t\t\t\t\trevision = EXCLUDED.revision,\n\t\t\t\t\t\tthrough_sequence = EXCLUDED.through_sequence,\n\t\t\t\t\t\tepoch = EXCLUDED.epoch,\n\t\t\t\t\t\tsource = EXCLUDED.source,\n\t\t\t\t\t\tsource_hash = EXCLUDED.source_hash,\n\t\t\t\t\t\tdocument = EXCLUDED.document,\n\t\t\t\t\t\tsidecar = EXCLUDED.sidecar,\n\t\t\t\t\t\tcreated_at = EXCLUDED.created_at\n\t\t\t\t`;\n\t\t\t\tawait transaction`\n\t\t\t\t\tDELETE FROM channel_updates WHERE channel_id = ${input.channelId}\n\t\t\t\t`;\n\t\t\t\tawait transaction`\n\t\t\t\t\tUPDATE channels\n\t\t\t\t\tSET revision = ${revision}, next_sequence = ${sequence + 1},\n\t\t\t\t\t\tupdated_at = GREATEST(updated_at, ${input.now})\n\t\t\t\t\tWHERE id = ${input.channelId}\n\t\t\t\t`;\n\t\t\t\treturn { revision, sequence, repeated: false };\n\t\t\t}));\n\t}\n\n\t#acquire(name: string, owner: string, ttlMs: number): Promise {\n\t\treturn this.#run(\"acquire storage lease\", async () => {\n\t\t\tlet [saved] = await this.#sql`\n\t\t\t\tINSERT INTO storage_leases (name, owner, fencing, expires_at)\n\t\t\t\tVALUES (\n\t\t\t\t\t${name}, ${owner}, 1,\n\t\t\t\t\tclock_timestamp() + (${ttlMs} * interval '1 millisecond')\n\t\t\t\t)\n\t\t\t\tON CONFLICT (name) DO UPDATE SET\n\t\t\t\t\towner = EXCLUDED.owner,\n\t\t\t\t\tfencing = CASE\n\t\t\t\t\t\tWHEN storage_leases.owner = EXCLUDED.owner\n\t\t\t\t\t\t\tAND storage_leases.expires_at > clock_timestamp()\n\t\t\t\t\t\tTHEN storage_leases.fencing\n\t\t\t\t\t\tELSE storage_leases.fencing + 1\n\t\t\t\t\tEND,\n\t\t\t\t\texpires_at = EXCLUDED.expires_at\n\t\t\t\tWHERE storage_leases.expires_at <= clock_timestamp()\n\t\t\t\t\tOR storage_leases.owner = EXCLUDED.owner\n\t\t\t\tRETURNING ${this.#sql.unsafe(LEASE_COLUMNS)}\n\t\t\t`;\n\t\t\treturn saved ? lease(saved) : undefined;\n\t\t});\n\t}\n\n\t#renew(held: Lease, ttlMs: number): Promise {\n\t\treturn this.#run(\"renew storage lease\", async () => {\n\t\t\tlet [saved] = await this.#sql`\n\t\t\t\tUPDATE storage_leases\n\t\t\t\tSET expires_at = clock_timestamp() + (${ttlMs} * interval '1 millisecond')\n\t\t\t\tWHERE name = ${held.name}\n\t\t\t\t\tAND owner = ${held.owner}\n\t\t\t\t\tAND fencing = ${held.fencing}\n\t\t\t\t\tAND expires_at > clock_timestamp()\n\t\t\t\tRETURNING ${this.#sql.unsafe(LEASE_COLUMNS)}\n\t\t\t`;\n\t\t\treturn saved ? lease(saved) : undefined;\n\t\t});\n\t}\n\n\t#release(held: Lease): Promise {\n\t\treturn this.#run(\"release storage lease\", async () => {\n\t\t\tlet released = await this.#sql<{ name: string }[]>`\n\t\t\t\tUPDATE storage_leases\n\t\t\t\tSET expires_at = to_timestamp(0)\n\t\t\t\tWHERE name = ${held.name} AND owner = ${held.owner} AND fencing = ${held.fencing}\n\t\t\t\tRETURNING name\n\t\t\t`;\n\t\t\treturn released.length > 0;\n\t\t});\n\t}\n\n\tasync #assertLease(transaction: TransactionSQL, held: Lease): Promise {\n\t\tlet [locked] = await transaction<{ name: string }[]>`\n\t\t\tSELECT name\n\t\t\tFROM storage_leases\n\t\t\tWHERE name = ${held.name}\n\t\t\tFOR UPDATE\n\t\t`;\n\t\tif (!locked) throw conflict(`storage lease ${held.name} is no longer held`);\n\t\tlet [active] = await transaction<{ name: string }[]>`\n\t\t\tSELECT name\n\t\t\tFROM storage_leases\n\t\t\tWHERE name = ${held.name}\n\t\t\t\tAND owner = ${held.owner}\n\t\t\t\tAND fencing = ${held.fencing}\n\t\t\t\tAND expires_at > clock_timestamp()\n\t\t`;\n\t\tif (!active) throw conflict(`storage lease ${held.name} is no longer held`);\n\t}\n}\n","editable",false,"branch","documentNavigation","repo","githubnext/chopin","repoId","7d1a8ac3-35b0-4dea-85d4-06a980f3bada","ownerId","org:89615882","site","nexthub","private","admin","settings","maintainer","triage","starred","stars",0,"watchers","watching","pinned","notify","views",[183,184,185,186,187,188,189,190,191,192,193,194,195],"navigation",{"_54":55,"_139":140,"_141":142,"_143":144,"_145":146,"_147":148,"_149":150},"sources",[],"viewer","meta",{"_131":132,"_133":134,"_135":136,"_42":31,"_137":126,"_138":31,"_48":49,"_50":49},"branches",[73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,116,117,118,119,120,121,122,123,124,125,126,127,128,129,130],"tags",[69,70,71,72],"cloneUrl","https://git.nexthub.sh/githubnext/chopin.git","color-controls-slice-1","color-controls-slice-2","color-controls-slice-3","color-controls-slice-4","Krzysztof-Cieslak-patch-1","Krzysztof-Cieslak-patch-2","Maggie/app-preview-toolkit","Maggie/ci-playwright-install-timeout","Maggie/collaborative-visual-decision","Maggie/design-audit","Maggie/generic-visual-decision-runtime","Maggie/impeccable-chopin-metadata","Maggie/jev-chat-product","Maggie/jev-composed-assessment","Maggie/native-document-layouts","Maggie/openui-composition-product","Maggie/openui-composition-trial","Maggie/planner-visual-authoring","Maggie/planner-visual-requests","Maggie/visual-selection-eval","artifacts/pr-419-live-preview-20261009","automation/pr-maintenance-state","automation/use-effect/callout-disabled-picker-8cf9e49ba5c09471","automation/use-effect/collaboration-options-ref-4e3a5a08a879a5ae","automation/use-effect/room-workspace-effective-access-a57d40866c71bed0","chore/bun-1.4.2","color-controls","color-controls-plan","comments","design-pr-assets","design/archived-decisions-badge","design/channel-route-remount","design/chat-decision-entry","design/conversation-analysis-markers","design/decision-evidence","design/settled-line-icon","dev/integration","diagram-step-controls","feat/liveapp-frontend-pilot","feat/local-experiments","feat/local-launcher-tracer","fix/autolink-mdx-crash","github-pills/2-pills","github-pills/3-hover-card","maggie/app-preview-toolkit","maggie/ci-playwright-install-timeout","maggie/collaborative-visual-decision","maggie/design-audit","maggie/generic-visual-decision-runtime","maggie/impeccable-chopin-metadata","maggie/jev-composed-assessment","maggie/native-document-layouts","maggie/openui-composition-product","maggie/openui-composition-trial","maggie/planner-visual-authoring","maggie/planner-visual-requests","maggie/visual-selection-eval","main","plain-text-fence-chrome","tokens/09-copy","wireframes/2-renderer","wireframes/3-planner","description","Let's compose rich plans together","owner","githubnext","section","","head","empty","pins",[165,166,167],"documents",{"_160":161,"_162":163},"code",{"_156":157,"_158":159},"automations",{"_153":154,"_155":-5},"delivery",{"_151":49,"_152":-5},"tag","04zvKQ8QOrDUF8TO1Ovz0Zy8_TT","releases","applications","workflows",3,"deployments","commits",1000,"changes",13,"counts",{"_164":163},"total",11,"issue",{"_168":158,"_170":181,"_15":158,"_172":182,"_52":174,"_175":31,"_176":174,"_177":159},{"_168":178,"_170":179,"_15":178,"_172":180,"_52":174,"_175":31,"_176":174,"_177":49},{"_168":169,"_170":171,"_15":169,"_172":173,"_52":174,"_175":31,"_176":174,"_177":163},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_168":158,"_170":181,"_15":158,"_172":226,"_52":174,"_175":31,"_176":174},{"_168":178,"_170":179,"_15":178,"_172":225,"_52":174,"_175":31,"_176":174},{"_168":169,"_170":171,"_15":169,"_172":224,"_52":174,"_175":31,"_176":174},{"_168":221,"_170":222,"_15":221,"_172":223,"_52":31,"_175":31,"_176":174},{"_168":218,"_170":219,"_15":218,"_172":220,"_52":31,"_175":31,"_176":174},{"_168":153,"_170":216,"_15":145,"_172":217,"_52":31,"_175":31,"_176":174},{"_168":155,"_170":213,"_15":214,"_172":215,"_52":31,"_175":31,"_176":174},{"_168":151,"_170":211,"_15":151,"_172":212,"_52":31,"_175":31,"_176":174},{"_168":207,"_170":208,"_15":209,"_172":210,"_52":31,"_175":31,"_176":174},{"_168":141,"_170":205,"_15":141,"_172":206,"_52":31,"_175":31,"_176":31},{"_168":202,"_170":203,"_15":202,"_172":204,"_52":31,"_175":31,"_176":31},{"_168":199,"_170":200,"_15":199,"_172":201,"_52":31,"_175":31,"_176":31},{"_168":196,"_170":197,"_15":196,"_172":198,"_52":31,"_175":31,"_176":31},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794486697823,"sidebar","expanded"]