From 84b18e4795e6d1cd9c6a8e7b8738cf988eff9f49 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:42:18 +0200 Subject: [PATCH 1/9] Fix double-encoded jsonb writes that trap the nostr publish loop Values bound to a Postgres ::jsonb parameter were passed through JSON.stringify first, so the driver encoded them a second time and the column stored a jsonb string scalar instead of an object. Every SQL path expression on those columns then evaluated to NULL, which made all three listSignedMissing* repair scans match unconditionally and reset published messages forever. Remove the manual serialization at all seven ::jsonb bind sites, backfill the two affected nostr_event columns idempotently on migration, and cap the retry via nostr_attempts, which was declared but never incremented. The same cap is mirrored in InMemoryMessageStore so the test double cannot disagree with the durable store. --- src/__tests__/lib/conversation-store.test.ts | 20 ++- src/__tests__/lib/message-store.test.ts | 140 ++++++++++++++++++- src/lib/conversation-store.ts | 6 +- src/lib/message-store.ts | 26 +++- 4 files changed, 176 insertions(+), 16 deletions(-) diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 405967a..4bf733b 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -74,13 +74,17 @@ function message(partial: Partial = {}): ConversationMes describe('CONVERSATION_SCHEMA_SQL', () => { it('creates conversation tables and unique indexes', () => { const joined = CONVERSATION_SCHEMA_SQL.join('\n'); - expect(CONVERSATION_SCHEMA_SQL).toHaveLength(8); + expect(CONVERSATION_SCHEMA_SQL).toHaveLength(9); expect(joined).toMatch(/CREATE TABLE IF NOT EXISTS conversation/i); expect(joined).toMatch(/CREATE TABLE IF NOT EXISTS conversation_message/i); expect(joined).toMatch(/conversation_member_member_uidx/); expect(joined).toMatch(/conversation_member_platform_uidx/); expect(joined).toMatch(/conversation_member_damus_uidx/); expect(joined).toMatch(/conversation_message_event_id_uidx/); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toBe( + `UPDATE conversation_message SET nostr_event = (nostr_event #>> '{}')::jsonb + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, + ); }); }); @@ -612,10 +616,19 @@ describe('PostgresConversationStore', () => { const row = message({ claimedUntil: 5_000, nostrEvent: { kind: 1059 } }); const created = await store.appendMessage(row); expect(sql.executes[0]?.text).toMatch(/INSERT INTO conversation_message/); + expect(typeof sql.executes[0]?.params[9]).not.toBe('string'); + expect(sql.executes[0]?.params[9]).toStrictEqual(row.nostrEvent); expect(sql.executes[1]?.text).toMatch(/UPDATE conversation SET last_message_at/); expect(created.text).toBe('hello'); }); + it('appendMessage binds a null nostrEvent unchanged', async () => { + const sql = new MockSql(); + const store = new PostgresConversationStore(sql); + await store.appendMessage(message()); + expect(sql.executes[0]?.params[9]).toBeNull(); + }); + it('appendMessage returns the existing row on event_id unique_violation', async () => { const sql = new MockSql(); sql.executeError = { code: '23505' }; @@ -660,9 +673,12 @@ describe('PostgresConversationStore', () => { it('updateSignedEvent returns false when no row matches', async () => { const sql = new MockSql(); + const nostrEvent = { k: 1 }; expect( - await new PostgresConversationStore(sql).updateSignedEvent('m', 'ab'.repeat(32), { k: 1 }), + await new PostgresConversationStore(sql).updateSignedEvent('m', 'ab'.repeat(32), nostrEvent), ).toBe(false); + expect(typeof sql.queries[0]?.params[2]).not.toBe('string'); + expect(sql.queries[0]?.params[2]).toStrictEqual(nostrEvent); }); it('updateSignedEvent returns false on unique_violation', async () => { diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index e59cc84..a2da2ad 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -109,6 +109,10 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.join('\n')).toMatch(/account_profile_message_id_fkey/); expect(MESSAGE_SCHEMA_SQL.join('\n')).toMatch(/ON DELETE SET NULL/); expect(MESSAGE_SCHEMA_SQL.join('\n')).toMatch(/account_profile_message_uidx/); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toBe( + `UPDATE message SET nostr_event = (nostr_event #>> '{}')::jsonb + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, + ); }); }); @@ -400,6 +404,115 @@ describe('InMemoryMessageStore', () => { expect((await store.getById('a'))?.eventId).toBe('cd'.repeat(32)); }); + it('listSignedMissingPhoto excludes rows at the publish-attempt cap', async () => { + const store = new InMemoryMessageStore([ + { + ...EARLY, + id: 'below-cap', + hasPhoto: true, + eventId: '11'.repeat(32), + nostrEvent: { content: '' }, + nostrPublishState: 'published', + nostrAttempts: 4, + }, + { + ...EARLY, + id: 'at-cap', + hasPhoto: true, + eventId: '22'.repeat(32), + nostrEvent: { content: '' }, + nostrPublishState: 'published', + nostrAttempts: 5, + }, + ]); + + expect((await store.listSignedMissingPhoto(10)).map((row) => row.id)).toEqual(['below-cap']); + }); + + it('listSignedMissingVideo excludes rows at the publish-attempt cap', async () => { + const store = new InMemoryMessageStore([ + { + ...EARLY, + id: 'below-cap', + hasVideo: true, + videoContentType: 'video/mp4', + eventId: '11'.repeat(32), + nostrEvent: { content: '' }, + nostrPublishState: 'published', + nostrAttempts: 4, + }, + { + ...EARLY, + id: 'at-cap', + hasVideo: true, + videoContentType: 'video/mp4', + eventId: '22'.repeat(32), + nostrEvent: { content: '' }, + nostrPublishState: 'published', + nostrAttempts: 5, + }, + ]); + + expect((await store.listSignedMissingVideo(10)).map((row) => row.id)).toEqual(['below-cap']); + }); + + it('listSignedMissingHashtags excludes rows at the publish-attempt cap', async () => { + const store = new InMemoryMessageStore([ + { + ...EARLY, + id: 'below-cap', + eventId: '11'.repeat(32), + nostrEvent: { content: 'missing hashtags' }, + nostrPublishState: 'published', + nostrAttempts: 4, + }, + { + ...EARLY, + id: 'at-cap', + eventId: '22'.repeat(32), + nostrEvent: { content: 'missing hashtags' }, + nostrPublishState: 'published', + nostrAttempts: 5, + }, + ]); + + expect((await store.listSignedMissingHashtags(10)).map((row) => row.id)).toEqual(['below-cap']); + }); + + it('resetSignedEvent counts attempts and preserves the first-attempt time', async () => { + const existingFirstAttemptAt = 1_234_567; + const store = new InMemoryMessageStore([ + { + ...EARLY, + id: 'without-first-attempt', + eventId: '11'.repeat(32), + nostrEvent: { content: '' }, + nostrPublishState: 'published', + nostrAttempts: 2, + nostrFirstAttemptAt: null, + }, + { + ...EARLY, + id: 'with-first-attempt', + eventId: '22'.repeat(32), + nostrEvent: { content: '' }, + nostrPublishState: 'published', + nostrAttempts: 3, + nostrFirstAttemptAt: existingFirstAttemptAt, + }, + ]); + + await store.resetSignedEvent('without-first-attempt', '11'.repeat(32)); + await store.resetSignedEvent('with-first-attempt', '22'.repeat(32)); + + const withoutFirstAttempt = await store.getById('without-first-attempt'); + expect(withoutFirstAttempt?.nostrAttempts).toBe(3); + expect(withoutFirstAttempt?.nostrFirstAttemptAt).toEqual(expect.any(Number)); + const withFirstAttempt = await store.getById('with-first-attempt'); + expect(withFirstAttempt?.nostrAttempts).toBe(4); + expect(withFirstAttempt?.nostrFirstAttemptAt).toBe(existingFirstAttemptAt); + }); + it('listSignedMissingPhoto and resetSignedEvent re-queue photo posts', async () => { const store = new InMemoryMessageStore(); const jpeg: ForumPhoto = { @@ -1087,7 +1200,7 @@ describe('PostgresMessageStore', () => { expect(sql.executes[0]?.params[5]).toBe('image/jpeg'); }); - it('create binds JSON-stringified nostrEvent when present', async () => { + it('create binds the nostrEvent object when present', async () => { const sql = new MockSql(); const store = new PostgresMessageStore(sql); const nostrEvent = { id: 'evt', kind: 1 }; @@ -1102,7 +1215,8 @@ describe('PostgresMessageStore', () => { nostrEvent, }; await store.create(row); - expect(sql.executes[0]?.params[13]).toBe(JSON.stringify(nostrEvent)); + expect(typeof sql.executes[0]?.params[13]).not.toBe('string'); + expect(sql.executes[0]?.params[13]).toStrictEqual(nostrEvent); }); it('create writes video bytes then binds video_content_type', async () => { @@ -1229,7 +1343,10 @@ describe('PostgresMessageStore', () => { expect(await store.claimUnsigned(5, 1_000, 60_000)).toEqual([]); expect(await store.claimUnpublished(5, 1_000, 60_000)).toEqual([]); expect(sql.queries.some((q) => /claimed_until <= \$2/.test(q.text))).toBe(true); - expect(await store.updateSignedEvent('m1', 'ee'.repeat(32), { id: 'x' })).toBe(false); + const nostrEvent = { id: 'x' }; + expect(await store.updateSignedEvent('m1', 'ee'.repeat(32), nostrEvent)).toBe(false); + expect(typeof sql.queries.at(-1)?.params[2]).not.toBe('string'); + expect(sql.queries.at(-1)?.params[2]).toStrictEqual(nostrEvent); await store.updatePublishState('m1', 'published', 'public'); await store.addSats('m1', 7); expect(sql.executes.some((e) => e.text.includes('sats = sats +'))).toBe(true); @@ -1424,12 +1541,18 @@ describe('PostgresMessageStore', () => { expect(listSql).toMatch(/photo IS NOT NULL/); expect(listSql).toMatch(/sats = 0/); expect(listSql).toMatch(/nostr_publish_state = 'published'/); + expect(listSql).toMatch(/nostr_attempts < 5/); expect(listSql).toMatch(/\/messages\/' \|\| id::text \|\| '\/photo\./); await store.resetSignedEvent('m1', 'ab'.repeat(32)); expect(sql.executes.at(-1)?.text).toMatch(/nostr_publish_state = 'pending'/); + expect(sql.executes.at(-1)?.text).toMatch(/nostr_attempts = message\.nostr_attempts \+ 1/); + expect(sql.executes.at(-1)?.text).toMatch( + /nostr_first_attempt_at = COALESCE\(message\.nostr_first_attempt_at, now\(\)\)/, + ); expect(sql.executes.at(-1)?.text).toMatch(/event_id IS NOT DISTINCT FROM/); expect(sql.executes.at(-1)?.text).toMatch(/sats = 0/); expect(sql.executes.at(-1)?.text).toMatch(/NOT EXISTS/); + expect(sql.executes.at(-1)?.params).toEqual(['m1', 'ab'.repeat(32)]); }); it('listSignedMissingVideo hits Postgres', async () => { @@ -1459,6 +1582,7 @@ describe('PostgresMessageStore', () => { ); expect(listSql).toMatch(/sats = 0/); expect(listSql).toMatch(/nostr_publish_state = 'published'/); + expect(listSql).toMatch(/nostr_attempts < 5/); expect(listSql).toMatch(/\/messages\/' \|\| id::text \|\| '\/video\./); }); @@ -1484,6 +1608,7 @@ describe('PostgresMessageStore', () => { expect(listSql).toMatch(/sats = 0/); expect(listSql).toMatch(/NOT EXISTS/); expect(listSql).toMatch(/nostr_publish_state = 'published'/); + expect(listSql).toMatch(/nostr_attempts < 5/); expect(listSql).toMatch(/jsonb_typeof\(nostr_event->'content'\) IS DISTINCT FROM 'string'/); expect(listSql).toContain('#21gifts([^a-z0-9_]|$)'); expect(listSql).toContain('#bitcoin([^a-z0-9_]|$)'); @@ -1548,9 +1673,11 @@ describe('PostgresMessageStore', () => { expect(sql.executes[0]?.text).toMatch(/INSERT INTO message_invoice/); expect(sql.executes[0]?.text).toMatch(/zap_request/); expect(sql.executes[0]?.text).toMatch(/lnurl_response/); - expect(sql.executes[0]?.params[7]).toBe(JSON.stringify({ kind: 9734 })); + expect(typeof sql.executes[0]?.params[7]).not.toBe('string'); + expect(sql.executes[0]?.params[7]).toStrictEqual(row.zapRequest); expect(sql.executes[0]?.params[14]).toBe(true); - expect(sql.executes[0]?.params[15]).toBe(JSON.stringify({ pr: 'lnbc1', status: 'OK' })); + expect(typeof sql.executes[0]?.params[15]).not.toBe('string'); + expect(sql.executes[0]?.params[15]).toStrictEqual(row.lnurlResponse); }); it('recordInvoiceAttempt binds null zap_request when the attempt has none', async () => { @@ -1714,7 +1841,8 @@ describe('PostgresMessageStore', () => { await store.recordZapIngest(row); expect(sql.executes).toHaveLength(1); expect(sql.executes[0]?.text).toMatch(/INSERT INTO nostr_zap_ingest/); - expect(sql.executes[0]?.params[9]).toBe(JSON.stringify({ id: 'r1', kind: 9735 })); + expect(typeof sql.executes[0]?.params[9]).not.toBe('string'); + expect(sql.executes[0]?.params[9]).toStrictEqual(row.receipt); }); it('listZapIngests maps receipt JSON string, non-indexed outcome, and null amount', async () => { diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index 3d83dcd..11c23f4 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -155,6 +155,8 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ `CREATE UNIQUE INDEX IF NOT EXISTS conversation_message_event_id_uidx ON conversation_message (event_id) WHERE event_id IS NOT NULL`, + `UPDATE conversation_message SET nostr_event = (nostr_event #>> '{}')::jsonb + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, ]; const THREAD_SELECT = `c.id, c.kind, c.account_a, c.account_b, c.counterpart_pubkey, c.created_at, c.last_message_at, @@ -678,7 +680,7 @@ export class PostgresConversationStore implements ConversationStore { row.name, row.eventId, row.nostrPublishState, - row.nostrEvent === null ? null : JSON.stringify(row.nostrEvent), + row.nostrEvent, row.claimedUntil === null ? null : new Date(row.claimedUntil), ], ); @@ -751,7 +753,7 @@ export class PostgresConversationStore implements ConversationStore { try { const rows = await this.#sql.query<{ id: string }>( `UPDATE conversation_message SET event_id = $2, nostr_event = $3::jsonb WHERE id = $1 RETURNING id`, - [id, eventId, JSON.stringify(nostrEvent)], + [id, eventId, nostrEvent], ); return rows[0] !== undefined; /* v8 ignore next 3 -- unique_violation on event_id */ diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index 89de20c..746155d 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -25,6 +25,8 @@ import { type ForumVideoContentType, } from '@/lib/video'; +const MAX_PUBLISH_ATTEMPTS = 5; + function kind1MissingPhotoUrl(event: Record | null, messageId: string): boolean { if (event === null) { return true; @@ -379,6 +381,8 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ FOREIGN KEY (profile_message_id) REFERENCES message (id) ON DELETE SET NULL`, `CREATE UNIQUE INDEX IF NOT EXISTS account_profile_message_uidx ON account (profile_message_id) WHERE profile_message_id IS NOT NULL`, + `UPDATE message SET nostr_event = (nostr_event #>> '{}')::jsonb + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, ]; /** @@ -668,6 +672,7 @@ export class InMemoryMessageStore implements MessageStore { row.hasVideo !== true && row.sats === 0 && row.nostrPublishState === 'published' && + row.nostrAttempts < MAX_PUBLISH_ATTEMPTS && !this.#rows.some((child) => child.parentId === row.id) && kind1MissingPhotoUrl(row.nostrEvent, row.id), ) @@ -691,6 +696,7 @@ export class InMemoryMessageStore implements MessageStore { row.videoContentType !== undefined && row.sats === 0 && row.nostrPublishState === 'published' && + row.nostrAttempts < MAX_PUBLISH_ATTEMPTS && !this.#rows.some((child) => child.parentId === row.id) && kind1MissingVideoUrl(row.nostrEvent, row.id), ) @@ -711,6 +717,7 @@ export class InMemoryMessageStore implements MessageStore { row.eventId !== null && row.sats === 0 && row.nostrPublishState === 'published' && + row.nostrAttempts < MAX_PUBLISH_ATTEMPTS && !this.#rows.some((child) => child.parentId === row.id) && kind1MissingHashtags(row.nostrEvent), ) @@ -735,6 +742,8 @@ export class InMemoryMessageStore implements MessageStore { row.nostrEvent = null; row.claimedUntil = null; row.nostrPublishState = 'pending'; + row.nostrAttempts += 1; + row.nostrFirstAttemptAt = row.nostrFirstAttemptAt ?? Date.now(); row.nostrPublishEpoch = null; } return Promise.resolve(); @@ -1084,7 +1093,7 @@ export class PostgresMessageStore implements MessageStore { stored.parentId, stored.authorPubkey, stored.eventId, - stored.nostrEvent === null ? null : JSON.stringify(stored.nostrEvent), + stored.nostrEvent, ], ); } catch (err) { @@ -1238,6 +1247,7 @@ export class PostgresMessageStore implements MessageStore { FROM message WHERE parent_id IS NULL AND event_id IS NOT NULL AND photo IS NOT NULL AND sats = 0 AND nostr_publish_state = 'published' + AND nostr_attempts < ${MAX_PUBLISH_ATTEMPTS} AND NOT EXISTS (SELECT 1 FROM message child WHERE child.parent_id = message.id) AND (video_content_type IS NULL OR video_content_type = '') AND ( @@ -1259,6 +1269,7 @@ export class PostgresMessageStore implements MessageStore { AND video_content_type IN ('video/mp4', 'video/webm', 'video/quicktime') AND sats = 0 AND nostr_publish_state = 'published' + AND nostr_attempts < ${MAX_PUBLISH_ATTEMPTS} AND NOT EXISTS (SELECT 1 FROM message child WHERE child.parent_id = message.id) AND ( nostr_event IS NULL @@ -1277,6 +1288,7 @@ export class PostgresMessageStore implements MessageStore { FROM message WHERE parent_id IS NULL AND event_id IS NOT NULL AND sats = 0 AND nostr_publish_state = 'published' + AND nostr_attempts < ${MAX_PUBLISH_ATTEMPTS} AND NOT EXISTS (SELECT 1 FROM message child WHERE child.parent_id = message.id) AND ( nostr_event IS NULL @@ -1294,7 +1306,9 @@ export class PostgresMessageStore implements MessageStore { async resetSignedEvent(id: string, expectedEventId: string | null): Promise { await this.#sql.execute( `UPDATE message SET event_id = NULL, nostr_event = NULL, claimed_until = NULL, - nostr_publish_state = 'pending', nostr_publish_epoch = NULL + nostr_publish_state = 'pending', nostr_publish_epoch = NULL, + nostr_attempts = message.nostr_attempts + 1, + nostr_first_attempt_at = COALESCE(message.nostr_first_attempt_at, now()) WHERE id = $1 AND event_id IS NOT DISTINCT FROM $2 AND sats = 0 AND NOT EXISTS (SELECT 1 FROM message child WHERE child.parent_id = message.id)`, [id, expectedEventId], @@ -1309,7 +1323,7 @@ export class PostgresMessageStore implements MessageStore { try { const rows = await this.#sql.query<{ id: string }>( `UPDATE message SET event_id = $2, nostr_event = $3::jsonb WHERE id = $1 RETURNING id`, - [id, eventId, JSON.stringify(nostrEvent)], + [id, eventId, nostrEvent], ); return rows[0] !== undefined; /* v8 ignore next 3 -- unique_violation on event_id */ @@ -1372,7 +1386,7 @@ export class PostgresMessageStore implements MessageStore { row.authorAccountId, row.amountSats, row.lightningAddress, - row.zapRequest === null ? null : JSON.stringify(row.zapRequest), + row.zapRequest, row.result, row.httpStatus, row.pr, @@ -1380,7 +1394,7 @@ export class PostgresMessageStore implements MessageStore { row.description, row.descriptionHash, row.isNip57Invoice, - row.lnurlResponse === null ? null : JSON.stringify(row.lnurlResponse), + row.lnurlResponse, ], ); } @@ -1417,7 +1431,7 @@ export class PostgresMessageStore implements MessageStore { row.reason, row.amountSats, row.receiptPubkey, - JSON.stringify(row.receipt), + row.receipt, ], ); } From 91f8fff0d04c4c2f24688b603a2ca7688a35d167 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:19:52 +0200 Subject: [PATCH 2/9] Document the one-time jsonb unwrap and the publish-attempt cap The schema arrays were still described as idempotent DDL although each now carries a one-time UPDATE, and the attempt cap added with the loop guard appeared in no TSDoc and no handbook section. Follow the db-change.ts precedent: name the DML in the array TSDoc, describe it in the schema mirror comment, and extend both handbook sections. --- docs/handbook/functions.md | 12 ++++++------ docs/schema/conversation.sql | 4 ++++ docs/schema/message.sql | 4 ++++ src/lib/conversation-store.ts | 2 +- src/lib/message-store.ts | 12 ++++++++++-- 5 files changed, 25 insertions(+), 9 deletions(-) diff --git a/docs/handbook/functions.md b/docs/handbook/functions.md index c1123de..6f6ed66 100644 --- a/docs/handbook/functions.md +++ b/docs/handbook/functions.md @@ -107,9 +107,9 @@ ## Function: migrateMessageSchema -- **Purpose:** Applies `MESSAGE_SCHEMA_SQL` in order (`CREATE TABLE IF NOT EXISTS message` with nullable `photo`/`photo_content_type`, newest-first index, additive `ALTER … ADD COLUMN IF NOT EXISTS` for existing databases including `video_content_type` (MIME in Postgres; video bytes on disk under `MEDIA_DIR`, not bytea), `parent_id uuid REFERENCES message (id)`, `author_pubkey text`, then `ALTER TABLE message ALTER COLUMN account_id DROP NOT NULL` and `CREATE INDEX IF NOT EXISTS message_parent_id_idx ON message (parent_id, created_at ASC, id ASC)`, then `message_invoice` and `nostr_zap_ingest` without FKs plus `ALTER TABLE message_invoice ADD COLUMN IF NOT EXISTS lnurl_response jsonb` and their `created_at`/`message_id` and `receipt_id` indexes). After `message` exists, adds `account_profile_message_id_fkey` (`ON DELETE SET NULL`) and unique partial index `account_profile_message_uidx`. +- **Purpose:** Applies `MESSAGE_SCHEMA_SQL` in order (`CREATE TABLE IF NOT EXISTS message` with nullable `photo`/`photo_content_type`, newest-first index, additive `ALTER … ADD COLUMN IF NOT EXISTS` for existing databases including `video_content_type` (MIME in Postgres; video bytes on disk under `MEDIA_DIR`, not bytea), `parent_id uuid REFERENCES message (id)`, `author_pubkey text`, then `ALTER TABLE message ALTER COLUMN account_id DROP NOT NULL` and `CREATE INDEX IF NOT EXISTS message_parent_id_idx ON message (parent_id, created_at ASC, id ASC)`, then `message_invoice` and `nostr_zap_ingest` without FKs plus `ALTER TABLE message_invoice ADD COLUMN IF NOT EXISTS lnurl_response jsonb` and their `created_at`/`message_id` and `receipt_id` indexes). After `message` exists, adds `account_profile_message_id_fkey` (`ON DELETE SET NULL`) and unique partial index `account_profile_message_uidx`. The array also runs a one-time, idempotent `UPDATE` unwrapping `nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`). - **Inputs:** `SqlClient`. -- **Returns / side effects:** Void; idempotent DDL execute matching `docs/schema/message.sql`. +- **Returns / side effects:** Void; idempotent SQL execute matching `docs/schema/message.sql`. - **Used by:** `openBootStores` when SQL opens. ## Function: migrateContactSchema @@ -121,9 +121,9 @@ ## Function: migrateConversationSchema -- **Purpose:** Applies `CONVERSATION_SCHEMA_SQL` in order (`conversation` + `conversation_message` tables and unique indexes). `db_change` attach runs later and covers the new public tables. +- **Purpose:** Applies `CONVERSATION_SCHEMA_SQL` in order (`conversation` + `conversation_message` tables and unique indexes), then runs a one-time, idempotent `UPDATE` unwrapping `conversation_message.nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`). `db_change` attach runs later and covers the new public tables. - **Inputs:** `SqlClient`. -- **Returns / side effects:** Void; idempotent DDL matching `docs/schema/conversation.sql`. +- **Returns / side effects:** Void; idempotent SQL matching `docs/schema/conversation.sql`. - **Used by:** `openBootStores` when SQL opens, after `migrateContactSchema` and before `migrateDbChangeSchema`. ## Function: migratePushSchema @@ -163,7 +163,7 @@ ## Function: PostgresMessageStore -- **Purpose:** Durable `MessageStore` over Postgres (`message` table plus `message_invoice` and `nostr_zap_ingest`). `deleteById` removes zap receipts, invoices, child replies, and the row in **one** parameterised data-modifying CTE `query`, then unlinks on-disk videos from the returned rows. `listLatest` is **top-level only** (`WHERE parent_id IS NULL`) with subquery `replyCount` (direct children), selecting Nostr columns plus `(photo IS NOT NULL) AS has_photo` and never the `photo` bytea column (HTTP window newest-first; product UX is a messenger group — clients reverse); `listReplies` is oldest-first (`WHERE parent_id = $1`, `created_at ASC, id ASC`); `listPublishedEventIds` returns non-null top-level `event_id`s newest-first for inbound reply REQ; `create` inserts optional photo bytes and optional `video_content_type` (disk write via `writeForumVideo`; `removeForumVideo` unlink on INSERT failure); `getPhoto` loads bytes by id; `getById`; `getByEventId` (`WHERE event_id`); `claimUnsigned`/`claimUnpublished` lease rows (`claimed_until <= now` is expired; unsigned requires `pending` + null `event_id`); `listPendingSigned` returns pending rows whose kind:1 lacks `t=bitcoin` (`created_at ASC, id ASC`); `clearSignedEvent` nulls `event_id` / `nostr_event` / `claimed_until` only while `pending` and `event_id` still matches the listed id and no child reply exists (`NOT EXISTS`); `listSignedMissingPhoto` returns published **top-level** rows (`parent_id IS NULL`) with a photo whose kind:1 content lacks `/messages/:id/photo.` plus an image extension (`sats = 0`, pending excluded so fan-out is not starved, video rows / `video_content_type` excluded so posters are not treated as missing photos, parents with children skipped via `NOT EXISTS`, `created_at ASC, id ASC`); `listSignedMissingVideo` returns published **top-level** rows (`parent_id IS NULL`) with `video_content_type` set whose kind:1 content lacks `/messages/:id/video.` (`sats = 0`, pending excluded, parents with children skipped via `NOT EXISTS`, `created_at ASC, id ASC`); `listSignedMissingHashtags` returns published unpaid **top-level** rows (`parent_id IS NULL`, parents with children skipped via `NOT EXISTS`) whose kind:1 content lacks a `#bitcoin` or `#21gifts` token (next character must not be `[A-Za-z0-9_]`; `sats = 0`, pending excluded so fan-out is not starved, includes null / non-string content, `created_at ASC, id ASC`); `resetSignedEvent` nulls `event_id` / `nostr_event` / `claimed_until`, parks `pending`, and clears the epoch only when `event_id` still matches, `sats` is 0, and no child reply exists (`NOT EXISTS`); `updateSignedEvent` (false on `event_id` collision); `updatePublishState`; `addSats`; `recordZapReceipt` (one statement: `INSERT nostr_zap_receipt ON CONFLICT DO NOTHING` plus `UPDATE message.sats`); `recordInvoiceAttempt` / `listInvoiceAttempts` (each attempt includes `lnurlResponse`: raw LNURL callback JSON object or null); `recordZapIngest` / `listZapIngests`. +- **Purpose:** Durable `MessageStore` over Postgres (`message` table plus `message_invoice` and `nostr_zap_ingest`). `deleteById` removes zap receipts, invoices, child replies, and the row in **one** parameterised data-modifying CTE `query`, then unlinks on-disk videos from the returned rows. `listLatest` is **top-level only** (`WHERE parent_id IS NULL`) with subquery `replyCount` (direct children), selecting Nostr columns plus `(photo IS NOT NULL) AS has_photo` and never the `photo` bytea column (HTTP window newest-first; product UX is a messenger group — clients reverse); `listReplies` is oldest-first (`WHERE parent_id = $1`, `created_at ASC, id ASC`); `listPublishedEventIds` returns non-null top-level `event_id`s newest-first for inbound reply REQ; `create` inserts optional photo bytes and optional `video_content_type` (disk write via `writeForumVideo`; `removeForumVideo` unlink on INSERT failure); `getPhoto` loads bytes by id; `getById`; `getByEventId` (`WHERE event_id`); `claimUnsigned`/`claimUnpublished` lease rows (`claimed_until <= now` is expired; unsigned requires `pending` + null `event_id`); `listPendingSigned` returns pending rows whose kind:1 lacks `t=bitcoin` (`created_at ASC, id ASC`); `clearSignedEvent` nulls `event_id` / `nostr_event` / `claimed_until` only while `pending` and `event_id` still matches the listed id and no child reply exists (`NOT EXISTS`); `listSignedMissingPhoto` returns published **top-level** rows (`parent_id IS NULL`) with a photo whose kind:1 content lacks `/messages/:id/photo.` plus an image extension (`sats = 0`, `nostr_attempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded so fan-out is not starved, video rows / `video_content_type` excluded so posters are not treated as missing photos, parents with children skipped via `NOT EXISTS`, `created_at ASC, id ASC`); `listSignedMissingVideo` returns published **top-level** rows (`parent_id IS NULL`) with `video_content_type` set whose kind:1 content lacks `/messages/:id/video.` (`sats = 0`, `nostr_attempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded, parents with children skipped via `NOT EXISTS`, `created_at ASC, id ASC`); `listSignedMissingHashtags` returns published unpaid **top-level** rows (`parent_id IS NULL`, parents with children skipped via `NOT EXISTS`) whose kind:1 content lacks a `#bitcoin` or `#21gifts` token (next character must not be `[A-Za-z0-9_]`; `sats = 0`, `nostr_attempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded so fan-out is not starved, includes null / non-string content, `created_at ASC, id ASC`); `resetSignedEvent` nulls `event_id` / `nostr_event` / `claimed_until`, parks `pending`, clears the epoch, increments `nostr_attempts`, and stamps `nostr_first_attempt_at` once, only when `event_id` still matches, `sats` is 0, and no child reply exists (`NOT EXISTS`); `updateSignedEvent` (false on `event_id` collision); `updatePublishState`; `addSats`; `recordZapReceipt` (one statement: `INSERT nostr_zap_receipt ON CONFLICT DO NOTHING` plus `UPDATE message.sats`); `recordInvoiceAttempt` / `listInvoiceAttempts` (each attempt includes `lnurlResponse`: raw LNURL callback JSON object or null); `recordZapIngest` / `listZapIngests`. - **Inputs:** Constructor takes a shared boot `SqlClient` (already migrated). - **Returns / side effects:** Parameter-bound SQL; maps snake_case rows to `MessageRow` / `ForumPhoto` / invoice and ingest rows. Claim uses `FOR UPDATE SKIP LOCKED`. Errors propagate to the route (503) except invoice/ingest persist failures which are caught by callers. - **Used by:** `openBootStores` when `DATABASE_URL` is set. @@ -387,7 +387,7 @@ ## Function: InMemoryMessageStore -- **Purpose:** Process-local `MessageStore` for the public member forum. Default empty so the process boots without a database. Photos live in a private map, not on listed rows. Same port as Postgres: `getById`, `deleteById` (row, direct replies, photos, invoices, zap receipt ids, on-disk videos), `getByEventId`, `listLatest` (top-level only, `parentId` null, each row has `replyCount`), `listReplies` (oldest-first for a parent), `listPublishedEventIds` (non-null top-level `eventId`s newest-first), claim/sign/publish (`claimUnsigned` is pending + null `eventId`; lease expires at `claimedUntil`), `listPendingSigned` (pending, no `t=bitcoin`, oldest-first), `clearSignedEvent` (pending and `eventId` still matches `expectedEventId` and the note has no child replies, then nulls `eventId` / `nostrEvent` / `claimedUntil`), `listSignedMissingPhoto` (top-level only, no children, published + photo, kind:1 content lacks `/messages/:id/photo.` plus extension, oldest-first, `sats === 0`, pending excluded, video rows excluded so posters are not treated as missing photos), `listSignedMissingVideo` (top-level only, no children, published + video MIME, kind:1 content lacks `/messages/:id/video.`, oldest-first, `sats === 0`, pending excluded), `listSignedMissingHashtags` (top-level only, no children, published unpaid, kind:1 content lacks a `#bitcoin` or `#21gifts` token, oldest-first, `sats === 0`, pending excluded so fan-out is not starved), `resetSignedEvent` (nulls `eventId` / `nostrEvent` / `claimedUntil`, parks `pending`, no-op unless `eventId` still matches, `sats` is 0, and the note has no child replies), `addSats`, `recordZapReceipt` (duplicate receipt id does not add sats; ids are released on `deleteById` so the same receipt can be recorded again), `recordInvoiceAttempt` / `listInvoiceAttempts` (each attempt includes `lnurlResponse` object or null), `recordZapIngest` / `listZapIngests`; `updateSignedEvent` returns false on duplicate `eventId`. Store/HTTP order is newest-first; product UX is a messenger group (clients reverse). +- **Purpose:** Process-local `MessageStore` for the public member forum. Default empty so the process boots without a database. Photos live in a private map, not on listed rows. Same port as Postgres: `getById`, `deleteById` (row, direct replies, photos, invoices, zap receipt ids, on-disk videos), `getByEventId`, `listLatest` (top-level only, `parentId` null, each row has `replyCount`), `listReplies` (oldest-first for a parent), `listPublishedEventIds` (non-null top-level `eventId`s newest-first), claim/sign/publish (`claimUnsigned` is pending + null `eventId`; lease expires at `claimedUntil`), `listPendingSigned` (pending, no `t=bitcoin`, oldest-first), `clearSignedEvent` (pending and `eventId` still matches `expectedEventId` and the note has no child replies, then nulls `eventId` / `nostrEvent` / `claimedUntil`), `listSignedMissingPhoto` (top-level only, no children, published + photo, kind:1 content lacks `/messages/:id/photo.` plus extension, oldest-first, `sats === 0`, `nostrAttempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded, video rows excluded so posters are not treated as missing photos), `listSignedMissingVideo` (top-level only, no children, published + video MIME, kind:1 content lacks `/messages/:id/video.`, oldest-first, `sats === 0`, `nostrAttempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded), `listSignedMissingHashtags` (top-level only, no children, published unpaid, kind:1 content lacks a `#bitcoin` or `#21gifts` token, oldest-first, `sats === 0`, `nostrAttempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded so fan-out is not starved), `resetSignedEvent` (nulls `eventId` / `nostrEvent` / `claimedUntil`, parks `pending`, clears `nostrPublishEpoch`, increments `nostrAttempts`, and stamps `nostrFirstAttemptAt` once, no-op unless `eventId` still matches, `sats` is 0, and the note has no child replies), `addSats`, `recordZapReceipt` (duplicate receipt id does not add sats; ids are released on `deleteById` so the same receipt can be recorded again), `recordInvoiceAttempt` / `listInvoiceAttempts` (each attempt includes `lnurlResponse` object or null), `recordZapIngest` / `listZapIngests`; `updateSignedEvent` returns false on duplicate `eventId`. Store/HTTP order is newest-first; product UX is a messenger group (clients reverse). - **Inputs:** Optional seed `MessageRow[]` (copied; `hasPhoto` defaults false). `listLatest(limit)` is top-level only (`parentId === null`) with `replyCount`, sorts newest `createdAt` then `id` DESC and caps at `limit`. `listReplies(parentId, limit?)` is oldest-first (default 200). `listPublishedEventIds(limit)` is newest-first non-null top-level `eventId`s. `create(row, photo?, video?)` appends a copy; `getPhoto(id)` returns a photo copy or null. - **Returns / side effects:** Promise of row/photo copies; mutating results does not change the store. Listed objects never expose bytes. When `video` is set, `create` awaits `writeForumVideo` (disk under `MEDIA_DIR`); if that write throws, the row is never pushed (no unlink). - **Used by:** `createApp` default `messageStore`. diff --git a/docs/schema/conversation.sql b/docs/schema/conversation.sql index 15192c0..9fd63fc 100644 --- a/docs/schema/conversation.sql +++ b/docs/schema/conversation.sql @@ -1,6 +1,10 @@ -- Private messaging threads and messages (member↔member, member↔platform, -- member↔Damus). Covered by db_change attach-all-public-tables. Plaintext is -- not a listed secret. Dedupe outbound/inbound by conversation_message.event_id. +-- migrateConversationSchema also runs a one-time, idempotent UPDATE unwrapping +-- conversation_message.nostr_event values stored as jsonb string scalars +-- (jsonb_typeof(nostr_event) = 'string'); that statement lives in the store's +-- CONVERSATION_SCHEMA_SQL array, not in this file. CREATE TABLE IF NOT EXISTS conversation ( id uuid PRIMARY KEY, diff --git a/docs/schema/message.sql b/docs/schema/message.sql index b226d8c..63f99c7 100644 --- a/docs/schema/message.sql +++ b/docs/schema/message.sql @@ -6,6 +6,10 @@ -- SELECT the photo column — use (photo IS NOT NULL) AS has_photo only. -- Optional video_content_type; bytes on disk under MEDIA_DIR (not bytea). -- ALTER ADD COLUMN IF NOT EXISTS keeps existing databases additive. +-- migrateMessageSchema also runs a one-time, idempotent UPDATE unwrapping +-- nostr_event values stored as jsonb string scalars +-- (jsonb_typeof(nostr_event) = 'string'); that statement lives in the store's +-- MESSAGE_SCHEMA_SQL array, not in this file. CREATE TABLE IF NOT EXISTS message ( id uuid PRIMARY KEY, diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index 11c23f4..42c7ecd 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -115,7 +115,7 @@ export interface ConversationStore { updatePublishState(id: string, state: NostrPublishState): Promise; } -/** Idempotent DDL for conversation tables (matches `docs/schema/conversation.sql`). */ +/** Idempotent SQL for conversation tables (DDL plus one-time unwrap of `nostr_event` values stored as jsonb string scalars in `conversation_message`; matches `docs/schema/conversation.sql`). */ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ `CREATE TABLE IF NOT EXISTS conversation ( id uuid PRIMARY KEY, diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index 746155d..80b317a 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -175,6 +175,8 @@ export interface MessageStore { * Parents that already have a child row are skipped for the same reason. * `sats = 0` only (zapped rows keep their event id). Pending rows are left * for fan-out — resetting them renews the sign lease and they never EVENT. + * Rows at or above `MAX_PUBLISH_ATTEMPTS` (5) are excluded so a row that can + * never satisfy a repair scan is not reset forever. * Oldest `createdAt` then `id` first. * * @param limit - Max rows. @@ -187,6 +189,8 @@ export interface MessageStore { * Parents that already have a child row are skipped for the same reason. * `sats = 0` only (zapped rows keep their event id). Pending rows are left * for fan-out — resetting them renews the sign lease and they never EVENT. + * Rows at or above `MAX_PUBLISH_ATTEMPTS` (5) are excluded so a row that can + * never satisfy a repair scan is not reset forever. * Oldest `createdAt` then `id` first. * * @param limit - Max rows. @@ -201,7 +205,9 @@ export interface MessageStore { * valid. `sats = 0` only (zapped rows keep * their event id). Pending rows are left for fan-out — resetting them * renews the sign lease and they never EVENT. Oldest `createdAt` then `id` - * first. Includes `nostrEvent === null` and non-string content. + * first. Rows at or above `MAX_PUBLISH_ATTEMPTS` (5) are excluded so a row + * that can never satisfy a repair scan is not reset forever. Includes + * `nostrEvent === null` and non-string content. * * @param limit - Max rows. */ @@ -211,6 +217,8 @@ export interface MessageStore { * Clear the signed event and park the row `pending` so it is signed again. * No-op unless `eventId` still matches `expectedEventId`, `sats` is 0, and * the note has no child replies. + * A successful reset increments `nostrAttempts` and stamps + * `nostrFirstAttemptAt` once when it is still unset. * * @param id - Message id. * @param expectedEventId - Event id observed when the row was listed. @@ -305,7 +313,7 @@ export interface ZapIngestRow { receipt: Record; } -/** Idempotent DDL for the forum table (matches `docs/schema/message.sql`). */ +/** Idempotent SQL for the forum table (DDL plus one-time unwrap of `nostr_event` values stored as jsonb string scalars; matches `docs/schema/message.sql`). */ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ `CREATE TABLE IF NOT EXISTS message ( id uuid PRIMARY KEY, From 875257e3a18844d15bf9d6f55b79cfdd5f335fb7 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:14:59 +0200 Subject: [PATCH 3/9] Make the jsonb backfill boot-safe, audited and self-clearing The backfill cast could abort the whole migration - and with it the process boot - on a single unparseable row, and it ran before migrateDbChangeSchema attaches trg_db_change, so its writes could land without an audit entry. Replace both statements with a DO block that returns early until the audit trigger is attached, repairs each row inside its own exception handler so one bad value is skipped with a warning instead of failing the boot, and clears nostr_attempts for the rows it repairs - the moment the root cause for that row actually disappears. The counter is deliberately not cleared on a successful publish: in this failure mode publishing always succeeds, so that would reset the cap every cycle and restore the unbounded loop. --- docs/handbook/functions.md | 5 +-- docs/schema/conversation.sql | 10 +++--- docs/schema/message.sql | 8 +++-- src/__tests__/lib/conversation-store.test.ts | 8 ++--- src/__tests__/lib/message-store.test.ts | 9 ++--- src/lib/conversation-store.ts | 35 ++++++++++++++++++-- src/lib/message-store.ts | 35 ++++++++++++++++++-- 7 files changed, 87 insertions(+), 23 deletions(-) diff --git a/docs/handbook/functions.md b/docs/handbook/functions.md index 6f6ed66..4b2b202 100644 --- a/docs/handbook/functions.md +++ b/docs/handbook/functions.md @@ -107,7 +107,7 @@ ## Function: migrateMessageSchema -- **Purpose:** Applies `MESSAGE_SCHEMA_SQL` in order (`CREATE TABLE IF NOT EXISTS message` with nullable `photo`/`photo_content_type`, newest-first index, additive `ALTER … ADD COLUMN IF NOT EXISTS` for existing databases including `video_content_type` (MIME in Postgres; video bytes on disk under `MEDIA_DIR`, not bytea), `parent_id uuid REFERENCES message (id)`, `author_pubkey text`, then `ALTER TABLE message ALTER COLUMN account_id DROP NOT NULL` and `CREATE INDEX IF NOT EXISTS message_parent_id_idx ON message (parent_id, created_at ASC, id ASC)`, then `message_invoice` and `nostr_zap_ingest` without FKs plus `ALTER TABLE message_invoice ADD COLUMN IF NOT EXISTS lnurl_response jsonb` and their `created_at`/`message_id` and `receipt_id` indexes). After `message` exists, adds `account_profile_message_id_fkey` (`ON DELETE SET NULL`) and unique partial index `account_profile_message_uidx`. The array also runs a one-time, idempotent `UPDATE` unwrapping `nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`). +- **Purpose:** Applies `MESSAGE_SCHEMA_SQL` in order (`CREATE TABLE IF NOT EXISTS message` with nullable `photo`/`photo_content_type`, newest-first index, additive `ALTER … ADD COLUMN IF NOT EXISTS` for existing databases including `video_content_type` (MIME in Postgres; video bytes on disk under `MEDIA_DIR`, not bytea), `parent_id uuid REFERENCES message (id)`, `author_pubkey text`, then `ALTER TABLE message ALTER COLUMN account_id DROP NOT NULL` and `CREATE INDEX IF NOT EXISTS message_parent_id_idx ON message (parent_id, created_at ASC, id ASC)`, then `message_invoice` and `nostr_zap_ingest` without FKs plus `ALTER TABLE message_invoice ADD COLUMN IF NOT EXISTS lnurl_response jsonb` and their `created_at`/`message_id` and `receipt_id` indexes). After `message` exists, adds `account_profile_message_id_fkey` (`ON DELETE SET NULL`) and unique partial index `account_profile_message_uidx`. On every boot, the array also runs an idempotent repair unwrapping `nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`), which matches no rows once complete. It is skipped while the `db_change` audit trigger is not attached and retried on the next boot; a row whose value cannot be parsed is skipped with a warning instead of failing the migration. Successfully repaired rows have `nostr_attempts` cleared for a fresh repair budget. - **Inputs:** `SqlClient`. - **Returns / side effects:** Void; idempotent SQL execute matching `docs/schema/message.sql`. - **Used by:** `openBootStores` when SQL opens. @@ -121,7 +121,7 @@ ## Function: migrateConversationSchema -- **Purpose:** Applies `CONVERSATION_SCHEMA_SQL` in order (`conversation` + `conversation_message` tables and unique indexes), then runs a one-time, idempotent `UPDATE` unwrapping `conversation_message.nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`). `db_change` attach runs later and covers the new public tables. +- **Purpose:** Applies `CONVERSATION_SCHEMA_SQL` in order (`conversation` + `conversation_message` tables and unique indexes), then runs an idempotent repair on every boot, unwrapping `conversation_message.nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`); it matches no rows once complete. The repair is skipped while the `db_change` audit trigger is not attached and retried on the next boot; a row whose value cannot be parsed is skipped with a warning instead of failing the migration. `db_change` attach runs later and covers the new public tables. - **Inputs:** `SqlClient`. - **Returns / side effects:** Void; idempotent SQL matching `docs/schema/conversation.sql`. - **Used by:** `openBootStores` when SQL opens, after `migrateContactSchema` and before `migrateDbChangeSchema`. @@ -164,6 +164,7 @@ ## Function: PostgresMessageStore - **Purpose:** Durable `MessageStore` over Postgres (`message` table plus `message_invoice` and `nostr_zap_ingest`). `deleteById` removes zap receipts, invoices, child replies, and the row in **one** parameterised data-modifying CTE `query`, then unlinks on-disk videos from the returned rows. `listLatest` is **top-level only** (`WHERE parent_id IS NULL`) with subquery `replyCount` (direct children), selecting Nostr columns plus `(photo IS NOT NULL) AS has_photo` and never the `photo` bytea column (HTTP window newest-first; product UX is a messenger group — clients reverse); `listReplies` is oldest-first (`WHERE parent_id = $1`, `created_at ASC, id ASC`); `listPublishedEventIds` returns non-null top-level `event_id`s newest-first for inbound reply REQ; `create` inserts optional photo bytes and optional `video_content_type` (disk write via `writeForumVideo`; `removeForumVideo` unlink on INSERT failure); `getPhoto` loads bytes by id; `getById`; `getByEventId` (`WHERE event_id`); `claimUnsigned`/`claimUnpublished` lease rows (`claimed_until <= now` is expired; unsigned requires `pending` + null `event_id`); `listPendingSigned` returns pending rows whose kind:1 lacks `t=bitcoin` (`created_at ASC, id ASC`); `clearSignedEvent` nulls `event_id` / `nostr_event` / `claimed_until` only while `pending` and `event_id` still matches the listed id and no child reply exists (`NOT EXISTS`); `listSignedMissingPhoto` returns published **top-level** rows (`parent_id IS NULL`) with a photo whose kind:1 content lacks `/messages/:id/photo.` plus an image extension (`sats = 0`, `nostr_attempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded so fan-out is not starved, video rows / `video_content_type` excluded so posters are not treated as missing photos, parents with children skipped via `NOT EXISTS`, `created_at ASC, id ASC`); `listSignedMissingVideo` returns published **top-level** rows (`parent_id IS NULL`) with `video_content_type` set whose kind:1 content lacks `/messages/:id/video.` (`sats = 0`, `nostr_attempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded, parents with children skipped via `NOT EXISTS`, `created_at ASC, id ASC`); `listSignedMissingHashtags` returns published unpaid **top-level** rows (`parent_id IS NULL`, parents with children skipped via `NOT EXISTS`) whose kind:1 content lacks a `#bitcoin` or `#21gifts` token (next character must not be `[A-Za-z0-9_]`; `sats = 0`, `nostr_attempts < MAX_PUBLISH_ATTEMPTS` (5, preventing a row that can never satisfy a repair scan from being reset forever), pending excluded so fan-out is not starved, includes null / non-string content, `created_at ASC, id ASC`); `resetSignedEvent` nulls `event_id` / `nostr_event` / `claimed_until`, parks `pending`, clears the epoch, increments `nostr_attempts`, and stamps `nostr_first_attempt_at` once, only when `event_id` still matches, `sats` is 0, and no child reply exists (`NOT EXISTS`); `updateSignedEvent` (false on `event_id` collision); `updatePublishState`; `addSats`; `recordZapReceipt` (one statement: `INSERT nostr_zap_receipt ON CONFLICT DO NOTHING` plus `UPDATE message.sats`); `recordInvoiceAttempt` / `listInvoiceAttempts` (each attempt includes `lnurlResponse`: raw LNURL callback JSON object or null); `recordZapIngest` / `listZapIngests`. +- **Backfill interaction:** The schema backfill clears `nostr_attempts` only for rows whose double-encoded `nostr_event` it successfully unwraps, because that repair removes the root cause and grants a fresh repair budget; successful publishing does not clear the cap. - **Inputs:** Constructor takes a shared boot `SqlClient` (already migrated). - **Returns / side effects:** Parameter-bound SQL; maps snake_case rows to `MessageRow` / `ForumPhoto` / invoice and ingest rows. Claim uses `FOR UPDATE SKIP LOCKED`. Errors propagate to the route (503) except invoice/ingest persist failures which are caught by callers. - **Used by:** `openBootStores` when `DATABASE_URL` is set. diff --git a/docs/schema/conversation.sql b/docs/schema/conversation.sql index 9fd63fc..efc98bd 100644 --- a/docs/schema/conversation.sql +++ b/docs/schema/conversation.sql @@ -1,10 +1,12 @@ -- Private messaging threads and messages (member↔member, member↔platform, -- member↔Damus). Covered by db_change attach-all-public-tables. Plaintext is -- not a listed secret. Dedupe outbound/inbound by conversation_message.event_id. --- migrateConversationSchema also runs a one-time, idempotent UPDATE unwrapping --- conversation_message.nostr_event values stored as jsonb string scalars --- (jsonb_typeof(nostr_event) = 'string'); that statement lives in the store's --- CONVERSATION_SCHEMA_SQL array, not in this file. +-- On every boot, migrateConversationSchema runs an idempotent repair unwrapping +-- conversation_message.nostr_event values stored as jsonb string scalars; it +-- matches no rows once complete. The repair is skipped until the db_change audit +-- trigger is attached and retried on the next boot. A value that cannot be parsed +-- is skipped with a warning instead of failing the migration. The statement +-- lives in the store's CONVERSATION_SCHEMA_SQL array, not in this file. CREATE TABLE IF NOT EXISTS conversation ( id uuid PRIMARY KEY, diff --git a/docs/schema/message.sql b/docs/schema/message.sql index 63f99c7..1556248 100644 --- a/docs/schema/message.sql +++ b/docs/schema/message.sql @@ -6,9 +6,11 @@ -- SELECT the photo column — use (photo IS NOT NULL) AS has_photo only. -- Optional video_content_type; bytes on disk under MEDIA_DIR (not bytea). -- ALTER ADD COLUMN IF NOT EXISTS keeps existing databases additive. --- migrateMessageSchema also runs a one-time, idempotent UPDATE unwrapping --- nostr_event values stored as jsonb string scalars --- (jsonb_typeof(nostr_event) = 'string'); that statement lives in the store's +-- On every boot, migrateMessageSchema runs an idempotent repair unwrapping +-- nostr_event values stored as jsonb string scalars; it matches no rows once +-- complete. The repair is skipped until the db_change audit trigger is attached +-- and retried on the next boot. A value that cannot be parsed is skipped with a +-- warning instead of failing the migration. The statement lives in the store's -- MESSAGE_SCHEMA_SQL array, not in this file. CREATE TABLE IF NOT EXISTS message ( diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 4bf733b..04bc758 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -81,10 +81,10 @@ describe('CONVERSATION_SCHEMA_SQL', () => { expect(joined).toMatch(/conversation_member_platform_uidx/); expect(joined).toMatch(/conversation_member_damus_uidx/); expect(joined).toMatch(/conversation_message_event_id_uidx/); - expect(CONVERSATION_SCHEMA_SQL.at(-1)).toBe( - `UPDATE conversation_message SET nostr_event = (nostr_event #>> '{}')::jsonb - WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, - ); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('FROM pg_trigger'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN others'); }); }); diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index a2da2ad..4259715 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -109,10 +109,11 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.join('\n')).toMatch(/account_profile_message_id_fkey/); expect(MESSAGE_SCHEMA_SQL.join('\n')).toMatch(/ON DELETE SET NULL/); expect(MESSAGE_SCHEMA_SQL.join('\n')).toMatch(/account_profile_message_uidx/); - expect(MESSAGE_SCHEMA_SQL.at(-1)).toBe( - `UPDATE message SET nostr_event = (nostr_event #>> '{}')::jsonb - WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, - ); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('FROM pg_trigger'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN others'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('nostr_attempts = 0'); }); }); diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index 42c7ecd..92e3cad 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -115,7 +115,7 @@ export interface ConversationStore { updatePublishState(id: string, state: NostrPublishState): Promise; } -/** Idempotent SQL for conversation tables (DDL plus one-time unwrap of `nostr_event` values stored as jsonb string scalars in `conversation_message`; matches `docs/schema/conversation.sql`). */ +/** Idempotent SQL for conversation tables (DDL plus boot-time unwrap of `nostr_event` values stored as jsonb string scalars in `conversation_message`; matches `docs/schema/conversation.sql`). */ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ `CREATE TABLE IF NOT EXISTS conversation ( id uuid PRIMARY KEY, @@ -155,8 +155,37 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ `CREATE UNIQUE INDEX IF NOT EXISTS conversation_message_event_id_uidx ON conversation_message (event_id) WHERE event_id IS NOT NULL`, - `UPDATE conversation_message SET nostr_event = (nostr_event #>> '{}')::jsonb - WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, + `DO $unwrap$ + DECLARE + repair_row RECORD; + BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_trigger + WHERE tgrelid = 'conversation_message'::regclass + AND tgname = 'trg_db_change' + AND NOT tgisinternal + ) THEN + RETURN; + END IF; + + FOR repair_row IN + SELECT id, nostr_event #>> '{}' AS unwrapped_event + FROM conversation_message + WHERE nostr_event IS NOT NULL + AND jsonb_typeof(nostr_event) = 'string' + LOOP + BEGIN + UPDATE conversation_message + SET nostr_event = repair_row.unwrapped_event::jsonb + WHERE id = repair_row.id; + EXCEPTION WHEN others THEN + RAISE WARNING 'Could not unwrap nostr_event for conversation_message id %', + repair_row.id; + END; + END LOOP; + END; + $unwrap$;`, ]; const THREAD_SELECT = `c.id, c.kind, c.account_a, c.account_b, c.counterpart_pubkey, c.created_at, c.last_message_at, diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index 80b317a..dd3e8d8 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -313,7 +313,7 @@ export interface ZapIngestRow { receipt: Record; } -/** Idempotent SQL for the forum table (DDL plus one-time unwrap of `nostr_event` values stored as jsonb string scalars; matches `docs/schema/message.sql`). */ +/** Idempotent SQL for the forum table (DDL plus boot-time unwrap of `nostr_event` values stored as jsonb string scalars; matches `docs/schema/message.sql`). */ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ `CREATE TABLE IF NOT EXISTS message ( id uuid PRIMARY KEY, @@ -389,8 +389,37 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ FOREIGN KEY (profile_message_id) REFERENCES message (id) ON DELETE SET NULL`, `CREATE UNIQUE INDEX IF NOT EXISTS account_profile_message_uidx ON account (profile_message_id) WHERE profile_message_id IS NOT NULL`, - `UPDATE message SET nostr_event = (nostr_event #>> '{}')::jsonb - WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, + `DO $unwrap$ + DECLARE + repair_row RECORD; + BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_trigger + WHERE tgrelid = 'message'::regclass + AND tgname = 'trg_db_change' + AND NOT tgisinternal + ) THEN + RETURN; + END IF; + + FOR repair_row IN + SELECT id, nostr_event #>> '{}' AS unwrapped_event + FROM message + WHERE nostr_event IS NOT NULL + AND jsonb_typeof(nostr_event) = 'string' + LOOP + BEGIN + UPDATE message + SET nostr_event = repair_row.unwrapped_event::jsonb, + nostr_attempts = 0 + WHERE id = repair_row.id; + EXCEPTION WHEN others THEN + RAISE WARNING 'Could not unwrap nostr_event for message id %', repair_row.id; + END; + END LOOP; + END; + $unwrap$;`, ]; /** From 8a98925dfc15e663a7619feab0a34ace954c4e5f Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:29:23 +0200 Subject: [PATCH 4/9] Match the backfill UPDATE against the live column, not the cursor snapshot PL/pgSQL fixes the cursor's MVCC snapshot when the loop opens, so the per-row UPDATE was writing a stale unwrapped value keyed only on id. A concurrent writer - an old replica still running resetSignedEvent, which nulls nostr_event - could have its newer state clobbered, and on message the forced nostr_attempts = 0 would re-arm the retry cap this PR installs. Select only the id, recompute the unwrap from the live column, and repeat the type predicate in the UPDATE's own WHERE so a since-changed row matches zero rows instead of being overwritten. --- src/__tests__/lib/conversation-store.test.ts | 7 +++++++ src/__tests__/lib/message-store.test.ts | 5 +++++ src/lib/conversation-store.ts | 8 +++++--- src/lib/message-store.ts | 8 +++++--- 4 files changed, 22 insertions(+), 6 deletions(-) diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 04bc758..8afdbd3 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -85,6 +85,13 @@ describe('CONVERSATION_SCHEMA_SQL', () => { expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN others'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain( + "SET nostr_event = (nostr_event #>> '{}')::jsonb", + ); + expect( + CONVERSATION_SCHEMA_SQL.at(-1)?.match(/jsonb_typeof\(nostr_event\) = 'string'/g), + ).toHaveLength(2); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); }); diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index 4259715..a22cfbb 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -114,6 +114,11 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN others'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('nostr_attempts = 0'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("SET nostr_event = (nostr_event #>> '{}')::jsonb"); + expect( + MESSAGE_SCHEMA_SQL.at(-1)?.match(/jsonb_typeof\(nostr_event\) = 'string'/g), + ).toHaveLength(2); + expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); }); diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index 92e3cad..cee4b17 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -170,15 +170,17 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ END IF; FOR repair_row IN - SELECT id, nostr_event #>> '{}' AS unwrapped_event + SELECT id FROM conversation_message WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string' LOOP BEGIN UPDATE conversation_message - SET nostr_event = repair_row.unwrapped_event::jsonb - WHERE id = repair_row.id; + SET nostr_event = (nostr_event #>> '{}')::jsonb + WHERE id = repair_row.id + AND nostr_event IS NOT NULL + AND jsonb_typeof(nostr_event) = 'string'; EXCEPTION WHEN others THEN RAISE WARNING 'Could not unwrap nostr_event for conversation_message id %', repair_row.id; diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index dd3e8d8..a2d95a7 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -404,16 +404,18 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ END IF; FOR repair_row IN - SELECT id, nostr_event #>> '{}' AS unwrapped_event + SELECT id FROM message WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string' LOOP BEGIN UPDATE message - SET nostr_event = repair_row.unwrapped_event::jsonb, + SET nostr_event = (nostr_event #>> '{}')::jsonb, nostr_attempts = 0 - WHERE id = repair_row.id; + WHERE id = repair_row.id + AND nostr_event IS NOT NULL + AND jsonb_typeof(nostr_event) = 'string'; EXCEPTION WHEN others THEN RAISE WARNING 'Could not unwrap nostr_event for message id %', repair_row.id; END; From 76ffd9cd24753f4ae7a453a383d1dfe2943df2aa Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:09:20 +0200 Subject: [PATCH 5/9] Narrow the repair exception, index its predicate, anchor its test EXCEPTION WHEN others swallowed every error class, so a permission or constraint failure during the repair would have been reduced to a warning and the boot would have continued as if nothing happened. Only the cast failure the handler was written for is tolerated now; anything else propagates. The repair predicate had no index, so every boot paid a sequential scan of message and conversation_message to establish that nothing was left to repair. A partial index over the same predicate is empty once the repair converges. The test counted occurrences of the type predicate, which stayed green even if both sat in the cursor query rather than one in the UPDATE's WHERE. It now anchors the predicate to the UPDATE statement. --- src/__tests__/lib/conversation-store.test.ts | 11 ++++++----- src/__tests__/lib/message-store.test.ts | 9 +++++---- src/lib/conversation-store.ts | 5 ++++- src/lib/message-store.ts | 5 ++++- 4 files changed, 19 insertions(+), 11 deletions(-) diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 8afdbd3..2025abc 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -74,7 +74,7 @@ function message(partial: Partial = {}): ConversationMes describe('CONVERSATION_SCHEMA_SQL', () => { it('creates conversation tables and unique indexes', () => { const joined = CONVERSATION_SCHEMA_SQL.join('\n'); - expect(CONVERSATION_SCHEMA_SQL).toHaveLength(9); + expect(CONVERSATION_SCHEMA_SQL).toHaveLength(10); expect(joined).toMatch(/CREATE TABLE IF NOT EXISTS conversation/i); expect(joined).toMatch(/CREATE TABLE IF NOT EXISTS conversation_message/i); expect(joined).toMatch(/conversation_member_member_uidx/); @@ -84,13 +84,14 @@ describe('CONVERSATION_SCHEMA_SQL', () => { expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('FROM pg_trigger'); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); - expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN others'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN others'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN invalid_text_representation'); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain( "SET nostr_event = (nostr_event #>> '{}')::jsonb", ); - expect( - CONVERSATION_SCHEMA_SQL.at(-1)?.match(/jsonb_typeof\(nostr_event\) = 'string'/g), - ).toHaveLength(2); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toMatch( + /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string'/, + ); expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); }); diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index a22cfbb..8e3a580 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -112,12 +112,13 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('FROM pg_trigger'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); - expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN others'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN others'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN invalid_text_representation'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('nostr_attempts = 0'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("SET nostr_event = (nostr_event #>> '{}')::jsonb"); - expect( - MESSAGE_SCHEMA_SQL.at(-1)?.match(/jsonb_typeof\(nostr_event\) = 'string'/g), - ).toHaveLength(2); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toMatch( + /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string'/, + ); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); }); diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index cee4b17..fff84cc 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -155,6 +155,9 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ `CREATE UNIQUE INDEX IF NOT EXISTS conversation_message_event_id_uidx ON conversation_message (event_id) WHERE event_id IS NOT NULL`, + `CREATE INDEX IF NOT EXISTS conversation_message_nostr_event_unrepaired_idx + ON conversation_message (id) + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, `DO $unwrap$ DECLARE repair_row RECORD; @@ -181,7 +184,7 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ WHERE id = repair_row.id AND nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'; - EXCEPTION WHEN others THEN + EXCEPTION WHEN invalid_text_representation THEN RAISE WARNING 'Could not unwrap nostr_event for conversation_message id %', repair_row.id; END; diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index a2d95a7..9ee93c2 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -389,6 +389,9 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ FOREIGN KEY (profile_message_id) REFERENCES message (id) ON DELETE SET NULL`, `CREATE UNIQUE INDEX IF NOT EXISTS account_profile_message_uidx ON account (profile_message_id) WHERE profile_message_id IS NOT NULL`, + `CREATE INDEX IF NOT EXISTS message_nostr_event_unrepaired_idx + ON message (id) + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'`, `DO $unwrap$ DECLARE repair_row RECORD; @@ -416,7 +419,7 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ WHERE id = repair_row.id AND nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'; - EXCEPTION WHEN others THEN + EXCEPTION WHEN invalid_text_representation THEN RAISE WARNING 'Could not unwrap nostr_event for message id %', repair_row.id; END; END LOOP; From 3ec882d76c275200a2fe7ea7bc543ae450cbedf8 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:24:12 +0200 Subject: [PATCH 6/9] Mirror the repair-predicate indexes into the schema docs The two partial indexes added with the boot repair existed only in the schema arrays, while the array TSDoc and the handbook both still claimed the mirrors matched. The mirror files carry every other index, and their existing disclaimer is scoped to the repair statement alone, so it does not cover DDL. Add both indexes to docs/schema and name them in the two migrate sections. --- docs/handbook/functions.md | 4 ++-- docs/schema/conversation.sql | 4 ++++ docs/schema/message.sql | 4 ++++ 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/docs/handbook/functions.md b/docs/handbook/functions.md index 4b2b202..73e671e 100644 --- a/docs/handbook/functions.md +++ b/docs/handbook/functions.md @@ -107,7 +107,7 @@ ## Function: migrateMessageSchema -- **Purpose:** Applies `MESSAGE_SCHEMA_SQL` in order (`CREATE TABLE IF NOT EXISTS message` with nullable `photo`/`photo_content_type`, newest-first index, additive `ALTER … ADD COLUMN IF NOT EXISTS` for existing databases including `video_content_type` (MIME in Postgres; video bytes on disk under `MEDIA_DIR`, not bytea), `parent_id uuid REFERENCES message (id)`, `author_pubkey text`, then `ALTER TABLE message ALTER COLUMN account_id DROP NOT NULL` and `CREATE INDEX IF NOT EXISTS message_parent_id_idx ON message (parent_id, created_at ASC, id ASC)`, then `message_invoice` and `nostr_zap_ingest` without FKs plus `ALTER TABLE message_invoice ADD COLUMN IF NOT EXISTS lnurl_response jsonb` and their `created_at`/`message_id` and `receipt_id` indexes). After `message` exists, adds `account_profile_message_id_fkey` (`ON DELETE SET NULL`) and unique partial index `account_profile_message_uidx`. On every boot, the array also runs an idempotent repair unwrapping `nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`), which matches no rows once complete. It is skipped while the `db_change` audit trigger is not attached and retried on the next boot; a row whose value cannot be parsed is skipped with a warning instead of failing the migration. Successfully repaired rows have `nostr_attempts` cleared for a fresh repair budget. +- **Purpose:** Applies `MESSAGE_SCHEMA_SQL` in order (`CREATE TABLE IF NOT EXISTS message` with nullable `photo`/`photo_content_type`, newest-first index, additive `ALTER … ADD COLUMN IF NOT EXISTS` for existing databases including `video_content_type` (MIME in Postgres; video bytes on disk under `MEDIA_DIR`, not bytea), `parent_id uuid REFERENCES message (id)`, `author_pubkey text`, then `ALTER TABLE message ALTER COLUMN account_id DROP NOT NULL` and `CREATE INDEX IF NOT EXISTS message_parent_id_idx ON message (parent_id, created_at ASC, id ASC)`, then `message_invoice` and `nostr_zap_ingest` without FKs plus `ALTER TABLE message_invoice ADD COLUMN IF NOT EXISTS lnurl_response jsonb` and their `created_at`/`message_id` and `receipt_id` indexes). After `message` exists, adds `account_profile_message_id_fkey` (`ON DELETE SET NULL`) and unique partial index `account_profile_message_uidx`. The partial index `message_nostr_event_unrepaired_idx` supports the boot repair's predicate so a converged table can be confirmed without a sequential scan. On every boot, the array also runs an idempotent repair unwrapping `nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`), which matches no rows once complete. It is skipped while the `db_change` audit trigger is not attached and retried on the next boot; a row whose value cannot be parsed is skipped with a warning instead of failing the migration. Successfully repaired rows have `nostr_attempts` cleared for a fresh repair budget. - **Inputs:** `SqlClient`. - **Returns / side effects:** Void; idempotent SQL execute matching `docs/schema/message.sql`. - **Used by:** `openBootStores` when SQL opens. @@ -121,7 +121,7 @@ ## Function: migrateConversationSchema -- **Purpose:** Applies `CONVERSATION_SCHEMA_SQL` in order (`conversation` + `conversation_message` tables and unique indexes), then runs an idempotent repair on every boot, unwrapping `conversation_message.nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`); it matches no rows once complete. The repair is skipped while the `db_change` audit trigger is not attached and retried on the next boot; a row whose value cannot be parsed is skipped with a warning instead of failing the migration. `db_change` attach runs later and covers the new public tables. +- **Purpose:** Applies `CONVERSATION_SCHEMA_SQL` in order (`conversation` + `conversation_message` tables and unique indexes). The partial index `conversation_message_nostr_event_unrepaired_idx` supports the boot repair's predicate so a converged table can be confirmed without a sequential scan. On every boot, the array runs an idempotent repair unwrapping `conversation_message.nostr_event` values stored as jsonb string scalars (`jsonb_typeof(nostr_event) = 'string'`); it matches no rows once complete. The repair is skipped while the `db_change` audit trigger is not attached and retried on the next boot; a row whose value cannot be parsed is skipped with a warning instead of failing the migration. `db_change` attach runs later and covers the new public tables. - **Inputs:** `SqlClient`. - **Returns / side effects:** Void; idempotent SQL matching `docs/schema/conversation.sql`. - **Used by:** `openBootStores` when SQL opens, after `migrateContactSchema` and before `migrateDbChangeSchema`. diff --git a/docs/schema/conversation.sql b/docs/schema/conversation.sql index efc98bd..3a3e2e8 100644 --- a/docs/schema/conversation.sql +++ b/docs/schema/conversation.sql @@ -47,3 +47,7 @@ CREATE INDEX IF NOT EXISTS conversation_message_conversation_id_idx CREATE UNIQUE INDEX IF NOT EXISTS conversation_message_event_id_uidx ON conversation_message (event_id) WHERE event_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS conversation_message_nostr_event_unrepaired_idx + ON conversation_message (id) + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'; diff --git a/docs/schema/message.sql b/docs/schema/message.sql index 1556248..ef4b009 100644 --- a/docs/schema/message.sql +++ b/docs/schema/message.sql @@ -95,3 +95,7 @@ ALTER TABLE account ADD CONSTRAINT account_profile_message_id_fkey FOREIGN KEY (profile_message_id) REFERENCES message (id) ON DELETE SET NULL; CREATE UNIQUE INDEX IF NOT EXISTS account_profile_message_uidx ON account (profile_message_id) WHERE profile_message_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS message_nostr_event_unrepaired_idx + ON message (id) + WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'; From ca7d548c2b8b35a94ddbd0c9f653326d459b4245 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 16:04:16 +0200 Subject: [PATCH 7/9] Catch the whole data-exception class in the repair loop Narrowing the handler to invalid_text_representation caught SQLSTATE 22P02 only. A nostr_event value carrying a NUL escape or a Unicode escape for a character absent from the database encoding raises 22P05 or 22021 instead - both in the same class, neither caught - so it would escape the block, abort the migration and abort the boot. The values come from arbitrary third-party clients, so those codes are reachable. Catch data_exception, the class category, so any bad value is skipped while permission errors, constraint violations and operational failures still propagate. The test now also asserts the single-code form is absent, so a future edit cannot silently narrow it again. --- src/__tests__/lib/conversation-store.test.ts | 7 +++++-- src/__tests__/lib/message-store.test.ts | 5 +++-- src/lib/conversation-store.ts | 2 +- src/lib/message-store.ts | 2 +- 4 files changed, 10 insertions(+), 6 deletions(-) diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 2025abc..5025bf9 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -85,12 +85,15 @@ describe('CONVERSATION_SCHEMA_SQL', () => { expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN others'); - expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN invalid_text_representation'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain( + 'EXCEPTION WHEN invalid_text_representation', + ); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception'); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain( "SET nostr_event = (nostr_event #>> '{}')::jsonb", ); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toMatch( - /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string'/, + /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string';/, ); expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index 8e3a580..06bc2b2 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -113,11 +113,12 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("tgname = 'trg_db_change'"); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN others'); - expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN invalid_text_representation'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN invalid_text_representation'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('nostr_attempts = 0'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("SET nostr_event = (nostr_event #>> '{}')::jsonb"); expect(MESSAGE_SCHEMA_SQL.at(-1)).toMatch( - /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string'/, + /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string';/, ); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index fff84cc..fa8d3fa 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -184,7 +184,7 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ WHERE id = repair_row.id AND nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'; - EXCEPTION WHEN invalid_text_representation THEN + EXCEPTION WHEN data_exception THEN RAISE WARNING 'Could not unwrap nostr_event for conversation_message id %', repair_row.id; END; diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index 9ee93c2..d3fe7bf 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -419,7 +419,7 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ WHERE id = repair_row.id AND nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string'; - EXCEPTION WHEN invalid_text_representation THEN + EXCEPTION WHEN data_exception THEN RAISE WARNING 'Could not unwrap nostr_event for message id %', repair_row.id; END; END LOOP; From d987c67758b409fe6a6095dd6f139ad3587b1639 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 16:34:03 +0200 Subject: [PATCH 8/9] Anchor the handler assertion on its trailing THEN toContain is a substring match, so EXCEPTION WHEN data_exception OR unique_violation THEN would have satisfied both the positive and the negative assertion and widened the handler undetected. Anchoring on the trailing THEN rejects any appended condition. --- src/__tests__/lib/conversation-store.test.ts | 2 +- src/__tests__/lib/message-store.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 5025bf9..0610cfa 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -88,7 +88,7 @@ describe('CONVERSATION_SCHEMA_SQL', () => { expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain( 'EXCEPTION WHEN invalid_text_representation', ); - expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception THEN'); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain( "SET nostr_event = (nostr_event #>> '{}')::jsonb", ); diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index 06bc2b2..467549f 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -114,7 +114,7 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN others'); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN invalid_text_representation'); - expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception THEN'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('nostr_attempts = 0'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("SET nostr_event = (nostr_event #>> '{}')::jsonb"); expect(MESSAGE_SCHEMA_SQL.at(-1)).toMatch( From adeebbf74c1d56e3b34cf331dfee3c345a13ae02 Mon Sep 17 00:00:00 2001 From: "David May (AI)" <323118616+davidleomayAI@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:13:29 +0200 Subject: [PATCH 9/9] Wrap only the cast in the repair handler, not the UPDATE The handler wrapped the whole UPDATE, and the UPDATE fires the db_change audit trigger - so it unavoidably covered the trigger too. That is why every attempt at a condition list came out either too broad (WHEN others, swallowing permission and operational failures) or too narrow (a single SQLSTATE, missing sibling codes and aborting the boot). Wrap only the cast. The trigger is no longer inside the handler, so an audit insert that fails on disk_full, insufficient_privilege or a sequence limit now propagates and aborts the boot, which is correct. data_exception on a bare text::jsonb cast is exactly the condition being handled, plus statement_too_complex, which is class 54 and reachable because tags: string[][] is a compile-time type only and inbound third-party events reach this column. The cast now reads the cursor snapshot, so the UPDATE carries a compare-and-swap on nostr_event to keep the guarantee an earlier round established. The warning is also truthful again: previously a trigger failure was reported as a value that could not be unwrapped. --- src/__tests__/lib/conversation-store.test.ts | 11 +++++++--- src/__tests__/lib/message-store.test.ts | 13 +++++++++--- src/lib/conversation-store.ts | 19 +++++++++++------- src/lib/message-store.ts | 21 ++++++++++++-------- 4 files changed, 43 insertions(+), 21 deletions(-) diff --git a/src/__tests__/lib/conversation-store.test.ts b/src/__tests__/lib/conversation-store.test.ts index 0610cfa..3f8174e 100644 --- a/src/__tests__/lib/conversation-store.test.ts +++ b/src/__tests__/lib/conversation-store.test.ts @@ -88,12 +88,17 @@ describe('CONVERSATION_SCHEMA_SQL', () => { expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain( 'EXCEPTION WHEN invalid_text_representation', ); - expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception THEN'); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain( - "SET nostr_event = (nostr_event #>> '{}')::jsonb", + 'EXCEPTION WHEN data_exception OR statement_too_complex THEN', ); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain( + "unwrapped := (repair_row.nostr_event #>> '{}')::jsonb;", + ); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('SET nostr_event = unwrapped'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('CONTINUE;'); + expect(CONVERSATION_SCHEMA_SQL.at(-1)).toContain('AND nostr_event = repair_row.nostr_event'); expect(CONVERSATION_SCHEMA_SQL.at(-1)).toMatch( - /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string';/, + /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string'[\s\S]*?AND nostr_event = repair_row\.nostr_event;/, ); expect(CONVERSATION_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); diff --git a/src/__tests__/lib/message-store.test.ts b/src/__tests__/lib/message-store.test.ts index 467549f..112b7be 100644 --- a/src/__tests__/lib/message-store.test.ts +++ b/src/__tests__/lib/message-store.test.ts @@ -114,11 +114,18 @@ describe('MESSAGE_SCHEMA_SQL', () => { expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("jsonb_typeof(nostr_event) = 'string'"); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN others'); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('EXCEPTION WHEN invalid_text_representation'); - expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('EXCEPTION WHEN data_exception THEN'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain( + 'EXCEPTION WHEN data_exception OR statement_too_complex THEN', + ); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain( + "unwrapped := (repair_row.nostr_event #>> '{}')::jsonb;", + ); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('SET nostr_event = unwrapped'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('nostr_attempts = 0'); - expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain("SET nostr_event = (nostr_event #>> '{}')::jsonb"); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('CONTINUE;'); + expect(MESSAGE_SCHEMA_SQL.at(-1)).toContain('AND nostr_event = repair_row.nostr_event'); expect(MESSAGE_SCHEMA_SQL.at(-1)).toMatch( - /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string';/, + /WHERE id = repair_row\.id[\s\S]*?jsonb_typeof\(nostr_event\) = 'string'[\s\S]*?AND nostr_event = repair_row\.nostr_event;/, ); expect(MESSAGE_SCHEMA_SQL.at(-1)).not.toContain('repair_row.unwrapped_event'); }); diff --git a/src/lib/conversation-store.ts b/src/lib/conversation-store.ts index fa8d3fa..8ebdbff 100644 --- a/src/lib/conversation-store.ts +++ b/src/lib/conversation-store.ts @@ -161,6 +161,7 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ `DO $unwrap$ DECLARE repair_row RECORD; + unwrapped jsonb; BEGIN IF NOT EXISTS ( SELECT 1 @@ -173,21 +174,25 @@ export const CONVERSATION_SCHEMA_SQL: readonly string[] = [ END IF; FOR repair_row IN - SELECT id + SELECT id, nostr_event FROM conversation_message WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string' LOOP BEGIN - UPDATE conversation_message - SET nostr_event = (nostr_event #>> '{}')::jsonb - WHERE id = repair_row.id - AND nostr_event IS NOT NULL - AND jsonb_typeof(nostr_event) = 'string'; - EXCEPTION WHEN data_exception THEN + unwrapped := (repair_row.nostr_event #>> '{}')::jsonb; + EXCEPTION WHEN data_exception OR statement_too_complex THEN RAISE WARNING 'Could not unwrap nostr_event for conversation_message id %', repair_row.id; + CONTINUE; END; + + UPDATE conversation_message + SET nostr_event = unwrapped + WHERE id = repair_row.id + AND nostr_event IS NOT NULL + AND jsonb_typeof(nostr_event) = 'string' + AND nostr_event = repair_row.nostr_event; END LOOP; END; $unwrap$;`, diff --git a/src/lib/message-store.ts b/src/lib/message-store.ts index d3fe7bf..5212897 100644 --- a/src/lib/message-store.ts +++ b/src/lib/message-store.ts @@ -395,6 +395,7 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ `DO $unwrap$ DECLARE repair_row RECORD; + unwrapped jsonb; BEGIN IF NOT EXISTS ( SELECT 1 @@ -407,21 +408,25 @@ export const MESSAGE_SCHEMA_SQL: readonly string[] = [ END IF; FOR repair_row IN - SELECT id + SELECT id, nostr_event FROM message WHERE nostr_event IS NOT NULL AND jsonb_typeof(nostr_event) = 'string' LOOP BEGIN - UPDATE message - SET nostr_event = (nostr_event #>> '{}')::jsonb, - nostr_attempts = 0 - WHERE id = repair_row.id - AND nostr_event IS NOT NULL - AND jsonb_typeof(nostr_event) = 'string'; - EXCEPTION WHEN data_exception THEN + unwrapped := (repair_row.nostr_event #>> '{}')::jsonb; + EXCEPTION WHEN data_exception OR statement_too_complex THEN RAISE WARNING 'Could not unwrap nostr_event for message id %', repair_row.id; + CONTINUE; END; + + UPDATE message + SET nostr_event = unwrapped, + nostr_attempts = 0 + WHERE id = repair_row.id + AND nostr_event IS NOT NULL + AND jsonb_typeof(nostr_event) = 'string' + AND nostr_event = repair_row.nostr_event; END LOOP; END; $unwrap$;`,