/**
 * The index store: every watched record, the links it names, and the cursors
 * that keep the stream resumable, in one SQLite database.
 *
 * The `appview_record` table holds each record whole as JSON. The
 * `appview_link` table is the reverse index: one row per at-uri a record
 * names, keyed by the JSON path it names it at, which is the whole question
 * a discovery port asks. Records and their links commit together, so a read
 * never sees half a write.
 *
 * The driver is the SQLite interface `@pdsjs/storage-sqlite/driver` names,
 * which better-sqlite3 and node:sqlite both implement natively.
 */

/** @typedef {import('@pdsjs/storage-sqlite/driver').SQLiteDatabase} SQLiteDatabase */
/** @typedef {import('@pdsjs/storage-sqlite/driver').SQLiteStatement} SQLiteStatement */

/**
 * @param {SQLiteDatabase} db
 */
export function createAppViewStore(db) {
  db.exec(`
    CREATE TABLE IF NOT EXISTS appview_record (
      uri TEXT PRIMARY KEY,
      did TEXT NOT NULL,
      collection TEXT NOT NULL,
      rkey TEXT NOT NULL,
      cid TEXT,
      value TEXT NOT NULL,
      indexed_at TEXT NOT NULL
    );
    CREATE INDEX IF NOT EXISTS idx_appview_record_collection
      ON appview_record (collection, indexed_at);
    CREATE TABLE IF NOT EXISTS appview_link (
      uri TEXT NOT NULL,
      path TEXT NOT NULL,
      subject TEXT NOT NULL,
      PRIMARY KEY (uri, path, subject)
    );
    CREATE INDEX IF NOT EXISTS idx_appview_link_subject
      ON appview_link (subject);
    CREATE TABLE IF NOT EXISTS appview_identity (
      did TEXT PRIMARY KEY,
      handle TEXT NOT NULL,
      updated_at TEXT NOT NULL
    );
    CREATE TABLE IF NOT EXISTS appview_cursor (
      source TEXT PRIMARY KEY,
      value TEXT NOT NULL
    );
  `);

  const now = () => new Date().toISOString();

  /**
   * @param {string} source
   * @returns {string|undefined}
   */
  const readCursor = (source) => {
    const row = /** @type {{value: string}|undefined} */ (
      db
        .prepare('SELECT value FROM appview_cursor WHERE source = ?')
        .get(source)
    );
    return row ? row.value : undefined;
  };

  /**
   * @param {string} source
   * @param {string} value
   */
  const writeCursor = (source, value) => {
    db.prepare(
      'INSERT OR REPLACE INTO appview_cursor (source, value) VALUES (?, ?)',
    ).run(source, value);
  };

  return {
    /**
     * @param {RecordWrite} write
     */
    putRecord(write) {
      const links = linksOf(write.value);
      transaction(db, () => {
        db.prepare(
          'INSERT OR REPLACE INTO appview_record (uri, did, collection, rkey, cid, value, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)',
        ).run(
          write.uri,
          write.did,
          write.collection,
          write.rkey,
          write.cid,
          JSON.stringify(write.value),
          now(),
        );
        db.prepare('DELETE FROM appview_link WHERE uri = ?').run(write.uri);
        for (const link of links) {
          db.prepare(
            'INSERT OR REPLACE INTO appview_link (uri, path, subject) VALUES (?, ?, ?)',
          ).run(write.uri, link.path, link.subject);
        }
      });
    },

    /**
     * @param {string} uri
     */
    deleteRecord(uri) {
      transaction(db, () => {
        db.prepare('DELETE FROM appview_record WHERE uri = ?').run(uri);
        db.prepare('DELETE FROM appview_link WHERE uri = ?').run(uri);
      });
    },

    /**
     * @param {string} uri
     * @returns {IndexedRecord|null}
     */
    getRecord(uri) {
      const row = /** @type {any|undefined} */ (
        db
          .prepare(
            'SELECT uri, did, collection, rkey, cid, value, indexed_at, rowid FROM appview_record WHERE uri = ?',
          )
          .get(uri)
      );
      return row ? toRecord(row) : null;
    },

    /**
     * Newest inserted first. `before` pages: only records inserted before
     * the rowid named. The order is the store's own insertion order, so a
     * page never revisits a row and never skips one.
     * @param {string} collection
     * @param {number} [limit]
     * @param {number} [before]
     * @returns {IndexedRecord[]}
     */
    listRecords(collection, limit = 100, before) {
      const rows = /** @type {any[]} */ (
        before
          ? db
              .prepare(
                'SELECT uri, did, collection, rkey, cid, value, indexed_at, rowid FROM appview_record WHERE collection = ? AND rowid < ? ORDER BY rowid DESC LIMIT ?',
              )
              .all(collection, before, limit)
          : db
              .prepare(
                'SELECT uri, did, collection, rkey, cid, value, indexed_at, rowid FROM appview_record WHERE collection = ? ORDER BY rowid DESC LIMIT ?',
              )
              .all(collection, limit)
      );
      return rows.map(toRecord);
    },

    /**
     * Every record of one collection naming a subject at a path. This is
     * the discovery port's whole question; the answer is each record's
     * address, read from its author when it matters.
     * @param {string} subjectUri
     * @param {string} collection
     * @param {string} [path]
     * @param {number} [limit]
     * @returns {IndexedRecord[]}
     */
    about(subjectUri, collection, path = 'subject.uri', limit = 1000) {
      const rows = /** @type {any[]} */ (
        db
          .prepare(
            `SELECT r.uri, r.did, r.collection, r.rkey, r.cid, r.value, r.indexed_at
             FROM appview_link l JOIN appview_record r ON r.uri = l.uri
             WHERE l.subject = ? AND l.path = ? AND r.collection = ?
             ORDER BY r.rowid DESC LIMIT ?`,
          )
          .all(subjectUri, path, collection, limit)
      );
      return rows.map(toRecord);
    },

    /**
     * @param {string} did
     * @param {string} handle
     */
    setIdentity(did, handle) {
      db.prepare(
        'INSERT OR REPLACE INTO appview_identity (did, handle, updated_at) VALUES (?, ?, ?)',
      ).run(did, handle, now());
    },

    /**
     * @param {string} did
     * @returns {string|null}
     */
    handleOf(did) {
      const row = /** @type {{handle: string}|undefined} */ (
        db.prepare('SELECT handle FROM appview_identity WHERE did = ?').get(did)
      );
      return row ? row.handle : null;
    },

    /**
     * An account deletion removes the account's records and their links.
     * @param {string} did
     */
    clearAccount(did) {
      transaction(db, () => {
        const uris = /** @type {any[]} */ (
          db.prepare('SELECT uri FROM appview_record WHERE did = ?').all(did)
        ).map((row) => String(row.uri));
        for (const uri of uris) {
          db.prepare('DELETE FROM appview_link WHERE uri = ?').run(uri);
        }
        db.prepare('DELETE FROM appview_record WHERE did = ?').run(did);
      });
    },

    /**
     * @param {string} source
     * @returns {number|undefined}
     */
    cursor(source) {
      const value = readCursor(source);
      if (value === undefined) return undefined;
      const seq = Number(value);
      return Number.isFinite(seq) ? seq : undefined;
    },

    /**
     * @param {string} source
     * @param {number} seq
     */
    setCursor(source, seq) {
      writeCursor(source, String(seq));
    },

    /**
     * @param {string} source
     * @returns {string|undefined}
     */
    textCursor(source) {
      return readCursor(source);
    },

    /**
     * @param {string} source
     * @param {string} value
     */
    setTextCursor(source, value) {
      writeCursor(source, value);
    },
  };
}

/** @typedef {ReturnType<typeof createAppViewStore>} AppViewStore */

/**
 * A record as the store holds it. `rowid` is the store's insertion order,
 * which is what `listRecords` pages by.
 * @typedef {Object} IndexedRecord
 * @property {string} uri
 * @property {string} did
 * @property {string} collection
 * @property {string} rkey
 * @property {string|null} cid
 * @property {Record<string, unknown>} value
 * @property {string} indexedAt
 * @property {number} rowid
 */

/**
 * A record write from any source: jetstream, the native firehose, or a
 * backfill pass. The cid is null for a row the archive replayed, because
 * the on-disk format does not store it.
 * @typedef {Object} RecordWrite
 * @property {string} did
 * @property {string} collection
 * @property {string} rkey
 * @property {string|null} cid
 * @property {string} uri
 * @property {number} seq
 * @property {Record<string, *>} value
 */

/**
 * A withdrawal: the record is gone from its author's repo.
 * @typedef {Object} RecordDelete
 * @property {string} did
 * @property {string} collection
 * @property {string} rkey
 * @property {string} uri
 * @property {number} seq
 * @property {true} deleted
 */

/**
 * The at-uri references one record names, as dotted JSON paths with array
 * positions folded away: `{subject: {uri: ...}}` indexes under
 * `subject.uri`, and a list of ref strings indexes under the list field's
 * name, one row per entry.
 * @param {unknown} value
 * @param {string} [path]
 * @returns {Array<{path: string, subject: string}>}
 */
export function linksOf(value, path = '') {
  if (typeof value === 'string') {
    return isRecordUri(value) ? [{ path, subject: value }] : [];
  }
  if (Array.isArray(value)) {
    /** @type {Array<{path: string, subject: string}>} */
    const found = [];
    for (const entry of value) found.push(...linksOf(entry, path));
    return found;
  }
  if (!value || typeof value !== 'object') return [];
  /** @type {Array<{path: string, subject: string}>} */
  const found = [];
  for (const [key, entry] of Object.entries(value)) {
    found.push(...linksOf(entry, path ? `${path}.${key}` : key));
  }
  return found;
}

/**
 * An at-uri that names a record: a DID, a collection and a record key.
 * A bare `at://did` names an account and indexes nothing.
 * @param {string} value
 * @returns {boolean}
 */
function isRecordUri(value) {
  if (!value.startsWith('at://')) return false;
  const segments = value.slice('at://'.length).split('/');
  return segments.length >= 3 && segments.every((part) => part.length > 0);
}

/**
 * @param {any} row
 * @returns {IndexedRecord}
 */
function toRecord(row) {
  return {
    uri: String(row.uri),
    did: String(row.did),
    collection: String(row.collection),
    rkey: String(row.rkey),
    cid: row.cid == null ? null : String(row.cid),
    value: JSON.parse(String(row.value)),
    indexedAt: String(row.indexed_at),
    rowid: Number(row.rowid),
  };
}

/**
 * @param {SQLiteDatabase} db
 * @param {() => void} work
 */
function transaction(db, work) {
  db.exec('BEGIN');
  try {
    work();
    db.exec('COMMIT');
  } catch (err) {
    db.exec('ROLLBACK');
    throw err;
  }
}