/**
 * Shared SQLite-backed storage for the P2P chat system.
 *
 * Drop-in replacement for ChatStore.  All Electron instances open the same
 * database file (appData/qortal-shared/chat.db).  WAL mode + busy_timeout
 * keep concurrent access safe without any extra coordination.
 *
 * Design notes
 * ─────────────
 * • No raw_json blob — columns are individually indexed and queried.
 * • syncState (authorAddress → contiguous seq) is maintained in-memory for
 *   O(1) hot-path access; rebuilt from DB at startup via a lightweight
 *   (chat_id, author_address, seq) query — no full event scan needed.
 * • readWatermarks are persisted to the DB so they survive restarts.
 * • Trim runs inside the same transaction as the triggering insert so the
 *   table never exceeds CHAT_MAX_EVENTS_PER_CHAT rows per chat.
 * • Bulk inserts (CHAT_SYNC_RESPONSE) use db.transaction() — one fsync.
 * • read_receipts rows are deleted in the same trim transaction as their
 *   parent chat_events rows — no orphaned receipts, no separate cleanup.
 * • getReadReceiptsForEvents uses query-scoped loading: receipts are fetched
 *   only for the event IDs currently held in renderer memory (bounded by
 *   viewport/history page size, not total message count).
 */

import Database, { type Database as DB, type Statement } from 'better-sqlite3';
import * as fs from 'fs';
import * as path from 'path';
import { log as loggerLog, error as loggerError } from './logger';
import type { ChatEvent, ChatSummary } from './chat';

// ── Constants (must mirror chat.ts) ──────────────────────────────────────────

const CHAT_MAX_EVENTS_PER_CHAT = 1_000;
const CHAT_MAX_SYNC_EVENTS = 200;

// ── Row shape returned from DB ────────────────────────────────────────────────

interface EventRow {
  id: string;
  chat_id: string;
  event_type: string;
  author_address: string;
  author_pub_key: string;
  seq: number;
  timestamp: number;
  content: string;
  reply_to: string | null;
  target_id: string | null;
  signature: string;
  attachment_meta: string | null;
  attachment_data_hash: string | null;
}

function rowToEvent(r: EventRow): ChatEvent {
  const ev: ChatEvent = {
    id: r.id,
    chatId: r.chat_id,
    eventType: r.event_type as ChatEvent['eventType'],
    authorAddress: r.author_address,
    authorPublicKey: r.author_pub_key,
    seq: r.seq,
    timestamp: r.timestamp,
    content: r.content,
    signature: r.signature,
  };
  if (r.reply_to != null) ev.replyTo = r.reply_to;
  if (r.target_id != null) ev.targetId = r.target_id;
  if (r.attachment_meta != null) {
    try {
      ev.attachmentMeta = JSON.parse(r.attachment_meta);
    } catch {
      // Malformed JSON — skip gracefully
    }
  }
  if (r.attachment_data_hash != null) ev.attachmentDataHash = r.attachment_data_hash;
  // attachmentData is intentionally NOT populated from history rows —
  // it lives in the chat_attachments table and is fetched on demand.
  return ev;
}

// ── ChatDatabase ──────────────────────────────────────────────────────────────

export class ChatDatabase {
  private db: DB;

  /** chatId → authorAddress → highest contiguous seq */
  private syncState = new Map<string, Map<string, number>>();
  /** chatId → read watermark timestamp */
  private readWatermarks = new Map<string, number>();
  /**
   * Per-instance dedup gate.  Mirrors what ChatStore's in-memory events array
   * used to do.  Populated at startup from existing DB rows so that events
   * already stored from a previous session are not re-emitted as "new".
   *
   * This is the critical difference from using info.changes from INSERT OR
   * IGNORE: with a shared DB another instance may have written an event first,
   * causing changes=0 for this instance even though this instance has never
   * seen the event and SHOULD emit it to the renderer.
   */
  private seenEventIds = new Set<string>();
  private seenEventIdToChatId = new Map<string, string>();

  // ── Prepared statements (compiled once, reused on every call) ─────────────
  private stmtInsert: Statement;
  private stmtCountForChat: Statement;
  private stmtTrimOldest: Statement;
  /**
   * Cascade-trim: deletes read_receipts whose event_id no longer exists in
   * chat_events for the given chat.  Run in the same transaction as
   * stmtTrimOldest so receipts are never orphaned.
   */
  private stmtTrimReceipts: Statement;
  private stmtGetEvents: Statement;
  private stmtGetEventsBefore: Statement;
  private stmtGetKnownChats: Statement;
  private stmtGetLastEvent: Statement;
  private stmtGetSeqsForSync: Statement;
  private stmtGetMissing: Statement;
  private stmtUpsertWatermark: Statement;
  private stmtLoadWatermarks: Statement;
  private stmtLoadSeqsForRebuild: Statement;
  /** Loads just event IDs for the seenEventIds Set — no content columns. */
  private stmtLoadEventIds: Statement;
  private stmtLoadEventIdsForChat: Statement;
  private stmtUpsertReceipt: Statement;
  private stmtGetReceiptsByReader: Statement;
  private stmtHasEvent: Statement;
  private stmtHasSeq: Statement;
  private stmtInsertAttachment: Statement;
  private stmtGetAttachment: Statement;
  private stmtTrimAttachments: Statement;
  private receiptLookupStatements = new Map<number, Statement>();

  constructor(dbPath: string) {
    fs.mkdirSync(path.dirname(dbPath), { recursive: true });

    this.db = new Database(dbPath);

    // Multi-process safety settings
    this.db.pragma('journal_mode = WAL');
    this.db.pragma('busy_timeout = 5000');
    this.db.pragma('synchronous = NORMAL');
    // Reduce write amplification
    this.db.pragma('wal_autocheckpoint = 1000');

    this.initSchema();
    this.stmtInsert = this.db.prepare(`
      INSERT OR IGNORE INTO chat_events
        (id, chat_id, event_type, author_address, author_pub_key,
         seq, timestamp, content, reply_to, target_id, signature,
         attachment_meta, attachment_data_hash)
      VALUES
        (@id, @chat_id, @event_type, @author_address, @author_pub_key,
         @seq, @timestamp, @content, @reply_to, @target_id, @signature,
         @attachment_meta, @attachment_data_hash)
    `);
    this.stmtCountForChat = this.db.prepare(
      'SELECT COUNT(*) AS cnt FROM chat_events WHERE chat_id = ?'
    );
    // Delete the oldest rows beyond the cap for this chat.
    // Uses the (chat_id, timestamp) index via the subquery.
    this.stmtTrimOldest = this.db.prepare(`
      DELETE FROM chat_events
      WHERE chat_id = ?
        AND id NOT IN (
          SELECT id FROM chat_events
          WHERE chat_id = ?
          ORDER BY timestamp DESC, seq DESC
          LIMIT ${CHAT_MAX_EVENTS_PER_CHAT}
        )
    `);
    // Cascade-trim receipts for events that were just purged from chat_events.
    // Runs in the same transaction as stmtTrimOldest.
    this.stmtTrimReceipts = this.db.prepare(`
      DELETE FROM read_receipts
      WHERE chat_id = ?
        AND event_id NOT IN (SELECT id FROM chat_events WHERE chat_id = ?)
    `);
    this.stmtGetEvents = this.db.prepare(`
      SELECT * FROM chat_events
      WHERE chat_id = ?
      ORDER BY timestamp ASC, seq ASC
      LIMIT ?
    `);
    this.stmtGetEventsBefore = this.db.prepare(`
      SELECT * FROM chat_events
      WHERE chat_id = ? AND timestamp < ?
      ORDER BY timestamp ASC, seq ASC
      LIMIT ?
    `);
    this.stmtGetKnownChats = this.db.prepare(
      'SELECT DISTINCT chat_id FROM chat_events'
    );
    // Last event per chat for summaries
    this.stmtGetLastEvent = this.db.prepare(`
      SELECT * FROM chat_events
      WHERE chat_id = ?
      ORDER BY timestamp DESC, seq DESC
      LIMIT 1
    `);
    // All (author_address, seq) rows for a chat — used by getSyncState rebuild
    this.stmtGetSeqsForSync = this.db.prepare(`
      SELECT author_address, seq
      FROM chat_events
      WHERE chat_id = ?
      ORDER BY author_address, seq
    `);
    // Events an author has beyond what the requester knows
    this.stmtGetMissing = this.db.prepare(`
      SELECT * FROM chat_events
      WHERE chat_id = ? AND author_address = ? AND seq > ?
      ORDER BY seq ASC
      LIMIT ?
    `);
    this.stmtUpsertWatermark = this.db.prepare(`
      INSERT INTO read_watermarks (chat_id, watermark)
      VALUES (?, ?)
      ON CONFLICT(chat_id) DO UPDATE SET watermark = excluded.watermark
        WHERE excluded.watermark > read_watermarks.watermark
    `);
    this.stmtLoadWatermarks = this.db.prepare(
      'SELECT chat_id, watermark FROM read_watermarks'
    );
    // Lightweight startup query — only (chat_id, author_address, seq), no content
    this.stmtLoadSeqsForRebuild = this.db.prepare(`
      SELECT chat_id, author_address, seq
      FROM chat_events
      ORDER BY chat_id, author_address, seq
    `);
    // Load IDs + chatIds so the in-memory dedup set can be trimmed with history.
    this.stmtLoadEventIds = this.db.prepare(
      'SELECT id, chat_id FROM chat_events'
    );
    this.stmtLoadEventIdsForChat = this.db.prepare(
      'SELECT id FROM chat_events WHERE chat_id = ?'
    );
    this.stmtUpsertReceipt = this.db.prepare(`
      INSERT OR IGNORE INTO read_receipts (chat_id, event_id, reader_address, read_at)
      VALUES (?, ?, ?, ?)
    `);
    this.stmtGetReceiptsByReader = this.db.prepare(`
      SELECT event_id FROM read_receipts
      WHERE chat_id = ? AND reader_address = ?
    `);
    this.stmtHasEvent = this.db.prepare(
      'SELECT 1 FROM chat_events WHERE id = ? LIMIT 1'
    );
    this.stmtHasSeq = this.db.prepare(
      'SELECT 1 FROM chat_events WHERE chat_id = ? AND author_address = ? AND seq = ? LIMIT 1'
    );
    this.stmtInsertAttachment = this.db.prepare(`
      INSERT OR IGNORE INTO chat_attachments (event_id, chat_id, data)
      VALUES (?, ?, ?)
    `);
    this.stmtGetAttachment = this.db.prepare(
      'SELECT data FROM chat_attachments WHERE event_id = ? LIMIT 1'
    );
    // Cascade-trim: removes attachment blobs for events purged from chat_events.
    // Runs in the same transaction as stmtTrimOldest so no orphaned blobs remain.
    this.stmtTrimAttachments = this.db.prepare(`
      DELETE FROM chat_attachments
      WHERE chat_id = ?
        AND event_id NOT IN (SELECT id FROM chat_events WHERE chat_id = ?)
    `);
  }

  private initSchema(): void {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS chat_events (
        id                  TEXT PRIMARY KEY,
        chat_id             TEXT NOT NULL,
        event_type          TEXT NOT NULL,
        author_address      TEXT NOT NULL,
        author_pub_key      TEXT NOT NULL,
        seq                 INTEGER NOT NULL,
        timestamp           INTEGER NOT NULL,
        content             TEXT NOT NULL,
        reply_to            TEXT,
        target_id           TEXT,
        signature           TEXT NOT NULL,
        attachment_meta     TEXT,
        attachment_data_hash TEXT
      );

      -- Primary read path: getEvents(chatId, limit, beforeTimestamp)
      CREATE INDEX IF NOT EXISTS idx_events_chat_ts
        ON chat_events(chat_id, timestamp);

      -- Sync path: getMissingEvents + getSyncState rebuild
      CREATE INDEX IF NOT EXISTS idx_events_chat_author_seq
        ON chat_events(chat_id, author_address, seq);

      CREATE TABLE IF NOT EXISTS read_watermarks (
        chat_id   TEXT PRIMARY KEY,
        watermark INTEGER NOT NULL
      );

      -- Per-message read receipts: who has seen which event.
      -- Trimmed in the same transaction as the parent chat_events rows.
      CREATE TABLE IF NOT EXISTS read_receipts (
        chat_id        TEXT    NOT NULL,
        event_id       TEXT    NOT NULL,
        reader_address TEXT    NOT NULL,
        read_at        INTEGER NOT NULL,
        PRIMARY KEY (chat_id, event_id, reader_address)
      );

      -- Query-scoped loading: WHERE event_id IN (...)
      CREATE INDEX IF NOT EXISTS idx_read_receipts_event
        ON read_receipts(event_id);

      -- Reconnect replay: all events read by a specific address in a chat
      CREATE INDEX IF NOT EXISTS idx_read_receipts_chat_reader
        ON read_receipts(chat_id, reader_address);

      -- Image attachment blobs stored separately from events so that all
      -- event queries remain lean (no accidental large blob loads).
      -- Trimmed in the same transaction as chat_events rows.
      CREATE TABLE IF NOT EXISTS chat_attachments (
        event_id TEXT PRIMARY KEY,
        chat_id  TEXT NOT NULL,
        data     TEXT NOT NULL
      );

      CREATE INDEX IF NOT EXISTS idx_attachments_chat
        ON chat_attachments(chat_id);

      -- Discovered peers — written by p2p-network.ts, schema defined here
      -- so the table exists whether or not p2p-network opens the DB first.
      CREATE TABLE IF NOT EXISTS discovered_peers (
        address       TEXT PRIMARY KEY,
        discovered_at INTEGER NOT NULL,
        source        TEXT NOT NULL
      );
    `);

    // Migration: add attachment columns to pre-existing databases that were
    // created before this schema version.
    this.runMigrations();
  }

  private runMigrations(): void {
    const existingCols = (
      this.db.prepare('PRAGMA table_info(chat_events)').all() as { name: string }[]
    ).map((r) => r.name);

    if (!existingCols.includes('attachment_meta')) {
      this.db.exec('ALTER TABLE chat_events ADD COLUMN attachment_meta TEXT');
    }
    if (!existingCols.includes('attachment_data_hash')) {
      this.db.exec('ALTER TABLE chat_events ADD COLUMN attachment_data_hash TEXT');
    }
  }

  /**
   * Rebuild in-memory caches from DB.
   * Kept async for API compatibility with ChatStore; implementation is sync.
   */
  async loadFromDisk(): Promise<void> {
    try {
      // Rebuild syncState from lightweight (chat_id, author_address, seq) scan
      const seqRows = this.stmtLoadSeqsForRebuild.all() as {
        chat_id: string;
        author_address: string;
        seq: number;
      }[];

      // Group seqs by (chatId, authorAddress)
      const grouped = new Map<string, Map<string, number[]>>();
      for (const row of seqRows) {
        let chatMap = grouped.get(row.chat_id);
        if (!chatMap) {
          chatMap = new Map();
          grouped.set(row.chat_id, chatMap);
        }
        let seqs = chatMap.get(row.author_address);
        if (!seqs) {
          seqs = [];
          chatMap.set(row.author_address, seqs);
        }
        seqs.push(row.seq); // rows come back ordered, no re-sort needed
      }

      for (const [chatId, authorMap] of grouped) {
        const syncMap = new Map<string, number>();
        for (const [author, seqs] of authorMap) {
          syncMap.set(author, computeContiguous(seqs));
        }
        this.syncState.set(chatId, syncMap);
      }

      // Restore watermarks
      const wmRows = this.stmtLoadWatermarks.all() as {
        chat_id: string;
        watermark: number;
      }[];
      for (const r of wmRows) {
        this.readWatermarks.set(r.chat_id, r.watermark);
      }

      const chatCount = grouped.size;
      const eventCount = seqRows.length;
      loggerLog(
        `[ChatDB] Loaded sync state for ${chatCount} chats (${eventCount} event seqs).`
      );

      // Populate the per-instance dedup set from all existing IDs so that
      // events already in the DB (from this or any other instance) are not
      // re-emitted as "new" when they arrive again via P2P sync.
      const idRows = this.stmtLoadEventIds.all() as {
        id: string;
        chat_id: string;
      }[];
      for (const r of idRows) {
        this.seenEventIds.add(r.id);
        this.seenEventIdToChatId.set(r.id, r.chat_id);
      }
      loggerLog(`[ChatDB] Seeded ${idRows.length} event IDs into dedup set.`);
    } catch (err) {
      loggerError('[ChatDB] Failed to rebuild caches:', err);
    }
  }

  /**
   * Insert a ChatEvent. Returns true if new to this instance, false if already seen.
   *
   * "New to this instance" is tracked via the per-instance seenEventIds Set,
   * NOT via info.changes from INSERT OR IGNORE.  This is critical for the
   * shared-DB multi-instance case: another instance may have already written
   * the event to the DB (info.changes = 0), but this instance still needs to
   * emit the event to its renderer.
   *
   * The DB insert is always attempted and is idempotent (INSERT OR IGNORE).
   * Trim runs only when the DB insert actually added a new row.
   */
  insert(event: ChatEvent): boolean {
    // Per-instance dedup: if we've already processed this event ID, skip.
    if (this.seenEventIds.has(event.id)) return false;
    this.seenEventIds.add(event.id);
    this.seenEventIdToChatId.set(event.id, event.chatId);

    const row = {
      id: event.id,
      chat_id: event.chatId,
      event_type: event.eventType,
      author_address: event.authorAddress,
      author_pub_key: event.authorPublicKey,
      seq: event.seq,
      timestamp: event.timestamp,
      content: event.content,
      reply_to: event.replyTo ?? null,
      target_id: event.targetId ?? null,
      signature: event.signature,
      attachment_meta: event.attachmentMeta ? JSON.stringify(event.attachmentMeta) : null,
      attachment_data_hash: event.attachmentDataHash ?? null,
    };

    // Persist idempotently. Trim only when this instance is the first to write
    // this event to the DB (info.changes > 0).
    const insertAndTrim = this.db.transaction(() => {
      const info = this.stmtInsert.run(row);
      if (info.changes > 0) {
        // Store attachment blob in the separate table if present.
        if (event.attachmentData) {
          this.stmtInsertAttachment.run(event.id, event.chatId, event.attachmentData);
        }
        const { cnt } = this.stmtCountForChat.get(event.chatId) as {
          cnt: number;
        };
        if (cnt > CHAT_MAX_EVENTS_PER_CHAT) {
          this.stmtTrimOldest.run(event.chatId, event.chatId);
          // Cascade: remove receipts and attachments for events that were just purged.
          this.stmtTrimReceipts.run(event.chatId, event.chatId);
          this.stmtTrimAttachments.run(event.chatId, event.chatId);
          this.refreshSeenEventIdsForChat(event.chatId);
        }
      }
    });

    insertAndTrim();
    this.updateSyncStateIncremental(event.chatId, event.authorAddress, event.seq);
    return true;
  }

  /**
   * Insert multiple events in a single transaction.
   * Returns the number of events that were new to this instance.
   */
  insertBatch(events: ChatEvent[]): number {
    if (events.length === 0) return 0;

    // Split into truly-new (not in seenEventIds) vs already-seen.
    const newEvents = events.filter((e) => !this.seenEventIds.has(e.id));
    if (newEvents.length === 0) return 0;

    // Mark all as seen before the DB work to prevent any race re-entry.
    for (const e of newEvents) {
      this.seenEventIds.add(e.id);
      this.seenEventIdToChatId.set(e.id, e.chatId);
    }

    const batchInsert = this.db.transaction(() => {
      for (const event of newEvents) {
        const row = {
          id: event.id,
          chat_id: event.chatId,
          event_type: event.eventType,
          author_address: event.authorAddress,
          author_pub_key: event.authorPublicKey,
          seq: event.seq,
          timestamp: event.timestamp,
          content: event.content,
          reply_to: event.replyTo ?? null,
          target_id: event.targetId ?? null,
          signature: event.signature,
          attachment_meta: event.attachmentMeta ? JSON.stringify(event.attachmentMeta) : null,
          attachment_data_hash: event.attachmentDataHash ?? null,
        };
        this.stmtInsert.run(row); // idempotent INSERT OR IGNORE
        if (event.attachmentData) {
          this.stmtInsertAttachment.run(event.id, event.chatId, event.attachmentData);
        }
        this.updateSyncStateIncremental(event.chatId, event.authorAddress, event.seq);
      }
      // Trim all affected chats once at the end
      const affectedChats = new Set(newEvents.map((e) => e.chatId));
      for (const chatId of affectedChats) {
        const { cnt } = this.stmtCountForChat.get(chatId) as { cnt: number };
        if (cnt > CHAT_MAX_EVENTS_PER_CHAT) {
          this.stmtTrimOldest.run(chatId, chatId);
          // Cascade: remove receipts and attachments for events that were just purged.
          this.stmtTrimReceipts.run(chatId, chatId);
          this.stmtTrimAttachments.run(chatId, chatId);
          this.refreshSeenEventIdsForChat(chatId);
        }
      }
    });
    batchInsert();
    return newEvents.length;
  }

  /**
   * Returns up to `limit` events for a chat, ordered oldest-first.
   * Optionally filters to events strictly before `beforeTimestamp`.
   */
  getEvents(chatId: string, limit = 50, beforeTimestamp?: number): ChatEvent[] {
    const rows =
      beforeTimestamp != null
        ? (this.stmtGetEventsBefore.all(chatId, beforeTimestamp, limit) as EventRow[])
        : (this.stmtGetEvents.all(chatId, limit) as EventRow[]);
    return rows.map(rowToEvent);
  }

  /**
   * Returns authorAddress → highest contiguous seq for the given chat.
   * Served from in-memory cache — O(1), no DB round-trip.
   */
  getSyncState(chatId: string): Record<string, number> {
    const m = this.syncState.get(chatId);
    return m ? Object.fromEntries(m.entries()) : {};
  }

  /** Returns all chatIds that have at least one stored event. */
  getKnownChatIds(): string[] {
    const rows = this.stmtGetKnownChats.all() as { chat_id: string }[];
    return rows.map((r) => r.chat_id);
  }

  /** Returns a summary for every known chat, sorted by most-recently-updated. */
  getChatSummaries(): ChatSummary[] {
    const chatIds = this.getKnownChatIds();
    const result: ChatSummary[] = [];
    for (const chatId of chatIds) {
      const lastRow = this.stmtGetLastEvent.get(chatId) as EventRow | undefined;
      const lastEvent = lastRow ? rowToEvent(lastRow) : null;
      const watermark = this.readWatermarks.get(chatId) ?? 0;

      // Count events newer than the watermark
      const unreadCount = (
        this.db
          .prepare(
            'SELECT COUNT(*) AS cnt FROM chat_events WHERE chat_id = ? AND timestamp > ?'
          )
          .get(chatId, watermark) as { cnt: number }
      ).cnt;

      result.push({
        chatId,
        lastEvent,
        unreadCount,
        updatedAt: lastEvent?.timestamp ?? 0,
      });
    }
    return result.sort((a, b) => b.updatedAt - a.updatedAt);
  }

  /**
   * Returns events this node has that the requester does not, capped at
   * CHAT_MAX_SYNC_EVENTS total across all authors.
   */
  getMissingEvents(
    chatId: string,
    theirSeqs: Record<string, number>
  ): ChatEvent[] {
    const missing: ChatEvent[] = [];
    const remaining = CHAT_MAX_SYNC_EVENTS;

    // Fetch missing per author; the index (chat_id, author_address, seq) serves this
    for (const [author, theirMax] of Object.entries(theirSeqs)) {
      if (missing.length >= remaining) break;
      const rows = this.stmtGetMissing.all(
        chatId,
        author,
        theirMax,
        remaining - missing.length
      ) as EventRow[];
      for (const r of rows) missing.push(rowToEvent(r));
    }

    // Also fetch from authors the requester hasn't mentioned at all
    if (missing.length < remaining) {
      const knownAuthors = new Set(Object.keys(theirSeqs));
      const allAuthors = (
        this.db
          .prepare(
            'SELECT DISTINCT author_address FROM chat_events WHERE chat_id = ?'
          )
          .all(chatId) as { author_address: string }[]
      ).map((r) => r.author_address);

      for (const author of allAuthors) {
        if (missing.length >= remaining) break;
        if (knownAuthors.has(author)) continue;
        const rows = this.stmtGetMissing.all(
          chatId,
          author,
          0, // they have nothing from this author
          remaining - missing.length
        ) as EventRow[];
        for (const r of rows) missing.push(rowToEvent(r));
      }
    }

    this.enrichWithAttachments(missing);
    return missing;
  }

  /**
   * For each event in the list that declares an attachment but has no blob yet,
   * look up the blob from chat_attachments and attach it inline.
   * Used so that P2P sync responses carry the full attachment data, allowing
   * offline peers to receive image blobs they missed during live broadcast.
   */
  private enrichWithAttachments(events: ChatEvent[]): void {
    for (const event of events) {
      if (event.attachmentMeta && !event.attachmentData) {
        const blob = this.getAttachment(event.id);
        if (blob) event.attachmentData = blob;
      }
    }
  }

  /**
   * Advances the read watermark for a chat.
   * Persists to DB so it survives restarts.
   */
  markRead(chatId: string, upToTimestamp: number): void {
    const current = this.readWatermarks.get(chatId) ?? 0;
    if (upToTimestamp <= current) return;
    this.readWatermarks.set(chatId, upToTimestamp);
    this.stmtUpsertWatermark.run(chatId, upToTimestamp);
  }

  /**
   * Record that `readerAddress` has seen `eventId` in `chatId`.
   * Idempotent — INSERT OR IGNORE means first read wins.
   */
  upsertReadReceipt(
    chatId: string,
    eventId: string,
    readerAddress: string,
    readAt: number
  ): void {
    this.stmtUpsertReceipt.run(chatId, eventId, readerAddress, readAt);
  }

  /**
   * Query-scoped receipt loading.
   *
   * Returns receipts only for the event IDs supplied — callers pass exactly
   * the IDs currently held in renderer memory (e.g. one history page), so
   * the result set is bounded by the viewport, not the total message count.
   *
   * Uses idx_read_receipts_event for an O(k log n) lookup.
   */
  getReadReceiptsForEvents(
    eventIds: string[]
  ): Record<string, string[]> {
    if (eventIds.length === 0) return {};
    const rows = this.getReceiptLookupStatement(eventIds.length).all(
      ...eventIds
    ) as { event_id: string; reader_address: string }[];

    const out: Record<string, string[]> = {};
    for (const r of rows) {
      if (!out[r.event_id]) out[r.event_id] = [];
      out[r.event_id].push(r.reader_address);
    }
    return out;
  }

  /**
   * Returns all event IDs in `chatId` that `readerAddress` has read.
   * Used for reconnect replay: after a peer reconnects, resend a CHAT_READ
   * envelope for events they authored that we've already seen.
   */
  getReadReceiptsByReader(chatId: string, readerAddress: string): string[] {
    const rows = this.stmtGetReceiptsByReader.all(
      chatId,
      readerAddress
    ) as { event_id: string }[];
    return rows.map((r) => r.event_id);
  }

  /**
   * Returns true when the event `id` exists in the local store.
   * Used to validate incoming CHAT_READ envelopes before persisting them.
   */
  hasEvent(id: string): boolean {
    return !!this.stmtHasEvent.get(id);
  }

  /**
   * Store an encrypted attachment blob for a given event.
   * Idempotent — INSERT OR IGNORE means first write wins.
   * Called from insert() / insertBatch() for events that carry attachmentData.
   * Also callable directly for deferred attachment storage.
   */
  insertAttachment(eventId: string, chatId: string, data: string): void {
    this.stmtInsertAttachment.run(eventId, chatId, data);
  }

  /**
   * Fetch the encrypted attachment blob for an event.
   * Returns null when no attachment exists (history event without data,
   * or the blob was never received).
   */
  getAttachment(eventId: string): string | null {
    const row = this.stmtGetAttachment.get(eventId) as { data: string } | undefined;
    return row?.data ?? null;
  }

  /**
   * No-op: SQLite writes are synchronous and durable.
   * Kept for API compatibility with ChatStore.
   */
  flushAllSync(): void {
    // Nothing to flush — every write already committed to WAL
  }

  /** Close the database connection. */
  stopAllTimers(): void {
    try {
      this.db.close();
    } catch {
      // Ignore close errors on shutdown
    }
  }

  // ── Internal sync-state helpers ───────────────────────────────────────────

  /**
   * Incrementally updates the contiguous seq for one author in one chat.
   * Called after every successful insert instead of recomputing the full chat.
   */
  private updateSyncStateIncremental(
    chatId: string,
    authorAddress: string,
    newSeq: number
  ): void {
    let chatMap = this.syncState.get(chatId);
    if (!chatMap) {
      chatMap = new Map();
      this.syncState.set(chatId, chatMap);
    }
    const current = chatMap.get(authorAddress) ?? 0;
    // Only extend the contiguous run if this seq is the next one
    if (newSeq === current + 1) {
      chatMap.set(authorAddress, newSeq);
      // Keep extending if we already have subsequent seqs in the DB
      this.extendContiguous(chatId, authorAddress, newSeq, chatMap);
    }
    // If newSeq > current+1 there's a gap — leave contiguous as-is;
    // if newSeq <= current it's a duplicate or already counted.
  }

  /**
   * After extending the contiguous seq to `from`, check whether the DB has
   * the next seq already (from a previously out-of-order insert) and keep
   * extending the run.
   */
  private extendContiguous(
    chatId: string,
    authorAddress: string,
    from: number,
    chatMap: Map<string, number>
  ): void {
    let next = from + 1;
    while (this.stmtHasSeq.get(chatId, authorAddress, next)) {
      chatMap.set(authorAddress, next);
      next++;
    }
  }

  private getReceiptLookupStatement(eventCount: number): Statement {
    let stmt = this.receiptLookupStatements.get(eventCount);
    if (stmt) return stmt;

    const placeholders = Array.from({ length: eventCount }, () => '?').join(', ');
    stmt = this.db.prepare(
      `SELECT event_id, reader_address FROM read_receipts WHERE event_id IN (${placeholders})`
    );
    this.receiptLookupStatements.set(eventCount, stmt);
    return stmt;
  }

  private refreshSeenEventIdsForChat(chatId: string): void {
    const retainedIds = new Set(
      (
        this.stmtLoadEventIdsForChat.all(chatId) as { id: string }[]
      ).map((row) => row.id)
    );

    for (const [eventId, mappedChatId] of this.seenEventIdToChatId.entries()) {
      if (mappedChatId !== chatId) continue;
      if (retainedIds.has(eventId)) continue;
      this.seenEventIdToChatId.delete(eventId);
      this.seenEventIds.delete(eventId);
    }

    for (const eventId of retainedIds) {
      this.seenEventIdToChatId.set(eventId, chatId);
      this.seenEventIds.add(eventId);
    }
  }
}

// ── Pure helper ───────────────────────────────────────────────────────────────

/**
 * Given a sorted array of seq numbers, return the highest value reachable
 * from the first element without any gap (contiguous prefix).
 * Matches the algorithm in ChatStore.updateSyncState.
 */
function computeContiguous(sortedSeqs: number[]): number {
  if (sortedSeqs.length === 0) return 0;
  // Treat anything before the earliest stored seq as already known
  let contiguous = sortedSeqs[0] - 1;
  for (const s of sortedSeqs) {
    if (s === contiguous + 1) {
      contiguous = s;
    } else {
      break;
    }
  }
  return contiguous;
}