// @pdsjs/storage-sqlite/space - SQLite SpaceStoragePort (proposal 0016).
//
// The schema is defined once in @pdsjs/core/space-schema and shared with the
// Cloudflare adapter; only the driver differs between them.

import {
  SPACE_MIGRATIONS,
  SPACE_SCHEMA_SQL,
  spaceRecordCursor,
  spaceRecordsQuery,
} from '@pdsjs/core/space-schema';

/** @typedef {{uri: string, space_did: string, space_type: string, is_owner: number, policy_json: string|null, created_at: string, deleted_at: string|null}} SpaceDbRow */
/** @typedef {{did: string, can_read: number, can_write: number}} SpaceMemberDbRow */
/** @typedef {{collection: string, rkey: string, cid: string, value: Uint8Array|Buffer, repo_rev: string, indexed_at: string}} SpaceRecordDbRow */
/** @typedef {Omit<SpaceRecordDbRow, 'value'> & {value?: Uint8Array|Buffer}} SpaceRecordListDbRow */
/** @typedef {{rev: string, idx: number, action: string, collection: string, rkey: string, cid: string|null, prev: string|null}} SpaceOpDbRow */
/** @typedef {{did: string, rev: string, data: Uint8Array|Buffer, space_rev: string}} SpaceWriterDbRow */
/** @typedef {{space: string, repo_rev: string, data: Uint8Array|Buffer, attempts: number, next_attempt_at: string, deadline_at: string}} SpaceNotifyDbRow */
/** @typedef {{space: string, service: string, service_endpoint: string, attempts: number, next_attempt_at: string, deadline_at: string}} SpaceDeletedNotifyDbRow */
/** @typedef {{service_did: string, service_endpoint: string, expires_at: string}} SpaceRecipientDbRow */

/**
 * @param {Uint8Array|Buffer|null} value
 * @returns {Uint8Array|null}
 */
function toBytes(value) {
  if (!value) return null;
  return value instanceof Uint8Array && value.constructor === Uint8Array
    ? value
    : new Uint8Array(value);
}

/**
 * @param {SpaceWriterDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceWriterRow}
 */
function toWriterRow(row) {
  return {
    did: row.did,
    repoRev: row.rev,
    data: /** @type {Uint8Array} */ (toBytes(row.data)),
    spaceRev: row.space_rev,
  };
}

/**
 * @param {SpaceNotifyDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceNotifyRow}
 */
function toNotifyRow(row) {
  return {
    space: row.space,
    repoRev: row.repo_rev,
    data: /** @type {Uint8Array} */ (toBytes(row.data)),
    attempts: row.attempts,
    nextAttemptAt: row.next_attempt_at,
    deadlineAt: row.deadline_at,
  };
}

/**
 * @param {SpaceDeletedNotifyDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceDeletedNotifyRow}
 */
function toDeletedNotifyRow(row) {
  return {
    space: row.space,
    service: row.service,
    serviceEndpoint: row.service_endpoint,
    attempts: row.attempts,
    nextAttemptAt: row.next_attempt_at,
    deadlineAt: row.deadline_at,
  };
}

/**
 * @param {SpaceDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceRow}
 */
function toSpaceRow(row) {
  return {
    uri: row.uri,
    spaceDid: row.space_did,
    spaceType: row.space_type,
    isOwner: row.is_owner === 1,
    policy: row.policy_json ? JSON.parse(row.policy_json) : null,
    createdAt: row.created_at,
    deletedAt: row.deleted_at,
  };
}

/**
 * @param {SpaceMemberDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceMemberRow}
 */
function toMemberRow(row) {
  return {
    did: row.did,
    read: row.can_read === 1,
    write: row.can_write === 1,
  };
}

/**
 * @param {SpaceRecordDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceRecordRow}
 */
function toRecordRow(row) {
  return {
    collection: row.collection,
    rkey: row.rkey,
    cid: row.cid,
    value: /** @type {Uint8Array} */ (toBytes(row.value)),
    repoRev: row.repo_rev,
    indexedAt: row.indexed_at,
  };
}

/**
 * The value column is absent from a metadata-only listing.
 * @param {SpaceRecordListDbRow} row
 * @returns {import('@pdsjs/core/ports').SpaceRecordListRow}
 */
function toRecordListRow(row) {
  return {
    collection: row.collection,
    rkey: row.rkey,
    cid: row.cid,
    ...(row.value === undefined
      ? {}
      : { value: /** @type {Uint8Array} */ (toBytes(row.value)) }),
    repoRev: row.repo_rev,
    indexedAt: row.indexed_at,
  };
}

/**
 * @param {import('./driver.js').SQLiteDatabase} db
 * @returns {import('@pdsjs/core/ports').SpaceStoragePort}
 */
export function createSpaceStorage(db) {
  db.exec(SPACE_SCHEMA_SQL);
  for (const statement of SPACE_MIGRATIONS) {
    try {
      db.exec(statement);
    } catch {
      // Already applied.
    }
  }

  const stmts = {
    getSpace: db.prepare('SELECT * FROM space WHERE uri = ?'),
    putSpace: db.prepare(
      `INSERT OR REPLACE INTO space
       (uri, space_did, space_type, is_owner, policy_json, created_at, deleted_at)
       VALUES (?, ?, ?, ?, ?, ?, ?)`,
    ),
    deleteSpace: db.prepare('UPDATE space SET deleted_at = ? WHERE uri = ?'),
    putMember: db.prepare(
      `INSERT INTO space_member (space, did, can_read, can_write) VALUES (?, ?, ?, ?)
       ON CONFLICT (space, did) DO UPDATE SET can_read = excluded.can_read, can_write = excluded.can_write`,
    ),
    removeMember: db.prepare(
      'DELETE FROM space_member WHERE space = ? AND did = ?',
    ),
    listMembers: db.prepare(
      'SELECT did, can_read, can_write FROM space_member WHERE space = ? AND did > ? ORDER BY did LIMIT ?',
    ),
    getMember: db.prepare(
      'SELECT can_read, can_write FROM space_member WHERE space = ? AND did = ?',
    ),
    removeAllMembers: db.prepare('DELETE FROM space_member WHERE space = ?'),
    removeWriter: db.prepare(
      'DELETE FROM space_writer WHERE space = ? AND did = ?',
    ),
    getRecord: db.prepare(
      'SELECT collection, rkey, cid, value, repo_rev, indexed_at FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
    ),
    getRecordCid: db.prepare(
      'SELECT cid FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
    ),
    listAllRecords: db.prepare(
      "SELECT collection, rkey, cid FROM space_record WHERE space = ? ORDER BY (collection || '/' || rkey)",
    ),
    putRecord: db.prepare(
      `INSERT OR REPLACE INTO space_record
       (space, collection, rkey, cid, value, repo_rev, indexed_at)
       VALUES (?, ?, ?, ?, ?, ?, ?)`,
    ),
    deleteRecord: db.prepare(
      'DELETE FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
    ),
    putOp: db.prepare(
      `INSERT INTO space_record_oplog
       (space, rev, idx, action, collection, rkey, cid, prev)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
    ),
    listOps: db.prepare(
      'SELECT rev, idx, action, collection, rkey, cid, prev FROM space_record_oplog WHERE space = ? AND rev > ? ORDER BY rev, idx LIMIT ?',
    ),
    putRepo: db.prepare(
      'INSERT OR REPLACE INTO space_repo (space, set_hash, rev) VALUES (?, ?, ?)',
    ),
    getRepo: db.prepare('SELECT set_hash, rev FROM space_repo WHERE space = ?'),
    getWriter: db.prepare(
      'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND did = ?',
    ),
    getSpaceRevHead: db.prepare(
      'SELECT MAX(space_rev) AS head FROM space_writer WHERE space = ?',
    ),
    putWriter: db.prepare(
      'INSERT OR REPLACE INTO space_writer (space, did, rev, data, space_rev) VALUES (?, ?, ?, ?, ?)',
    ),
    listWriters: db.prepare(
      'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND space_rev > ? ORDER BY space_rev LIMIT ?',
    ),
    putRevocation: db.prepare(
      `INSERT INTO space_credential_revocation (space, jti, expires_at)
       VALUES (?, ?, ?)
       ON CONFLICT (space, jti) DO UPDATE SET expires_at = MAX(expires_at, excluded.expires_at)`,
    ),
    deleteExpiredRevocations: db.prepare(
      'DELETE FROM space_credential_revocation WHERE expires_at <= ?',
    ),
    getRevocation: db.prepare(
      'SELECT 1 FROM space_credential_revocation WHERE space = ? AND jti = ? AND expires_at > ?',
    ),
    enqueueNotify: db.prepare(
      `INSERT OR REPLACE INTO space_notify_outbox
       (space, repo_rev, data, attempts, next_attempt_at, deadline_at)
       VALUES (?, ?, ?, 0, ?, ?)`,
    ),
    listDueNotify: db.prepare(
      'SELECT space, repo_rev, data, attempts, next_attempt_at, deadline_at FROM space_notify_outbox WHERE next_attempt_at <= ? ORDER BY next_attempt_at LIMIT ?',
    ),
    rescheduleNotify: db.prepare(
      'UPDATE space_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND repo_rev = ?',
    ),
    deleteNotify: db.prepare(
      'DELETE FROM space_notify_outbox WHERE space = ? AND repo_rev = ?',
    ),
    nextNotifyAt: db.prepare(
      `SELECT MIN(next) AS next FROM (
         SELECT MIN(next_attempt_at) AS next FROM space_notify_outbox
         UNION ALL
         SELECT MIN(next_attempt_at) AS next FROM space_deleted_notify_outbox
       )`,
    ),
    enqueueDeletedNotify: db.prepare(
      `INSERT OR REPLACE INTO space_deleted_notify_outbox
       (space, service, service_endpoint, attempts, next_attempt_at, deadline_at)
       VALUES (?, ?, ?, ?, ?, ?)`,
    ),
    listDueDeletedNotify: db.prepare(
      'SELECT space, service, service_endpoint, attempts, next_attempt_at, deadline_at FROM space_deleted_notify_outbox WHERE next_attempt_at <= ? ORDER BY next_attempt_at LIMIT ?',
    ),
    rescheduleDeletedNotify: db.prepare(
      'UPDATE space_deleted_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND service = ?',
    ),
    deleteDeletedNotify: db.prepare(
      'DELETE FROM space_deleted_notify_outbox WHERE space = ? AND service = ?',
    ),
    putRecipient: db.prepare(
      `INSERT OR REPLACE INTO space_credential_recipient
       (space, service_did, service_endpoint, expires_at)
       VALUES (?, ?, ?, ?)`,
    ),
    deleteRecipient: db.prepare(
      'DELETE FROM space_credential_recipient WHERE space = ? AND service_did = ?',
    ),
    listRecipients: db.prepare(
      'SELECT service_did, service_endpoint, expires_at FROM space_credential_recipient WHERE space = ? AND expires_at > ? ORDER BY service_did',
    ),
  };

  return {
    async getSpace(uri) {
      const row = /** @type {SpaceDbRow|undefined} */ (stmts.getSpace.get(uri));
      return row ? toSpaceRow(row) : null;
    },

    async putSpace(space) {
      stmts.putSpace.run(
        space.uri,
        space.spaceDid,
        space.spaceType,
        space.isOwner ? 1 : 0,
        space.policy ? JSON.stringify(space.policy) : null,
        space.createdAt,
        space.deletedAt,
      );
    },

    async deleteSpace(uri, deletedAt) {
      stmts.deleteSpace.run(deletedAt, uri);
    },

    async listSpaces({
      type = null,
      did = null,
      cursor = null,
      limit = 50,
      excludeDeleted = false,
    }) {
      // Filtering is a plain WHERE clause because the domain stores the URI's
      // authority and type as columns; the adapter never parses space URIs.
      // That also means LIMIT applies to matches, so no scan-and-refill loop.
      const where = ['uri > ?'];
      const params = /** @type {any[]} */ ([cursor ?? '']);
      if (did) {
        where.push('space_did = ?');
        params.push(did);
      }
      if (type) {
        where.push('space_type = ?');
        params.push(type);
      }
      if (excludeDeleted) where.push('deleted_at IS NULL');
      params.push(limit);
      const rows = /** @type {SpaceDbRow[]} */ (
        db
          .prepare(
            `SELECT * FROM space WHERE ${where.join(' AND ')} ORDER BY uri LIMIT ?`,
          )
          .all(...params)
      );
      const spaces = rows.map(toSpaceRow);
      return {
        spaces,
        // A full page always yields a cursor, matching every other paginated
        // method here and the memory adapter's `page` helper.
        cursor: spaces.length === limit ? spaces[spaces.length - 1].uri : null,
      };
    },

    async putMember(space, did, access) {
      stmts.putMember.run(
        space,
        did,
        access.read ? 1 : 0,
        access.write ? 1 : 0,
      );
    },

    async removeMember(space, did) {
      stmts.removeMember.run(space, did);
    },

    async listMembers(space, cursor, limit) {
      const rows = /** @type {SpaceMemberDbRow[]} */ (
        stmts.listMembers.all(space, cursor ?? '', limit)
      );
      const members = rows.map(toMemberRow);
      return {
        members,
        cursor:
          members.length === limit ? members[members.length - 1].did : null,
      };
    },

    async getMember(space, did) {
      const row = /** @type {SpaceMemberDbRow|undefined} */ (
        stmts.getMember.get(space, did)
      );
      return row
        ? { read: row.can_read === 1, write: row.can_write === 1 }
        : null;
    },

    async removeAllMembers(space) {
      stmts.removeAllMembers.run(space);
    },

    async getSpaceRecord(space, collection, rkey) {
      const row = /** @type {SpaceRecordDbRow|undefined} */ (
        stmts.getRecord.get(space, collection, rkey)
      );
      return row ? toRecordRow(row) : null;
    },

    async getSpaceRecordCid(space, collection, rkey) {
      const row = /** @type {{cid: string}|undefined} */ (
        stmts.getRecordCid.get(space, collection, rkey)
      );
      return row ? row.cid : null;
    },

    async listSpaceRecords(space, opts) {
      const { sql, params } = spaceRecordsQuery(space, opts);
      const rows = /** @type {SpaceRecordListDbRow[]} */ (
        db.prepare(sql).all(...params)
      );
      const records = rows.map(toRecordListRow);
      return {
        records,
        cursor:
          records.length === opts.limit
            ? spaceRecordCursor(records[records.length - 1])
            : null,
      };
    },

    async listAllSpaceRecords(space) {
      return /** @type {Array<{collection: string, rkey: string, cid: string}>} */ (
        stmts.listAllRecords.all(space)
      );
    },

    async commitSpaceWrite(space, { writes, setHash, rev, indexedAt }) {
      for (const w of writes) {
        if (w.action !== 'delete' && !(w.value instanceof Uint8Array)) {
          throw new Error(
            `space write value must be Uint8Array: ${w.collection}/${w.rkey}`,
          );
        }
      }

      db.exec('BEGIN');
      try {
        writes.forEach((w, idx) => {
          if (w.action === 'delete') {
            stmts.deleteRecord.run(space, w.collection, w.rkey);
          } else {
            stmts.putRecord.run(
              space,
              w.collection,
              w.rkey,
              w.cid,
              w.value,
              rev,
              indexedAt,
            );
          }
          stmts.putOp.run(
            space,
            rev,
            idx,
            w.action,
            w.collection,
            w.rkey,
            w.cid,
            w.prev,
          );
        });
        stmts.putRepo.run(space, setHash, rev);
        db.exec('COMMIT');
      } catch (err) {
        // Never let a rollback failure replace the real error.
        try {
          db.exec('ROLLBACK');
        } catch {
          // transaction already unwound by SQLite
        }
        throw err;
      }
    },

    async getSpaceRepo(space) {
      const row =
        /** @type {{set_hash: Uint8Array|Buffer|null, rev: string|null}|undefined} */ (
          stmts.getRepo.get(space)
        );
      if (!row) return null;
      return { setHash: toBytes(row.set_hash), rev: row.rev };
    },

    async listSpaceOps(space, since, limit) {
      const rows = /** @type {SpaceOpDbRow[]} */ (
        stmts.listOps.all(space, since ?? '', limit)
      );
      return rows.map((r) => ({
        rev: r.rev,
        idx: r.idx,
        action: /** @type {'create'|'update'|'delete'} */ (r.action),
        collection: r.collection,
        rkey: r.rkey,
        cid: r.cid,
        prev: r.prev,
      }));
    },

    async getSpaceWriter(space, did) {
      const row = /** @type {SpaceWriterDbRow|undefined} */ (
        stmts.getWriter.get(space, did)
      );
      return row ? toWriterRow(row) : null;
    },

    async getSpaceRevHead(space) {
      const row = /** @type {{head: string|null}|undefined} */ (
        stmts.getSpaceRevHead.get(space)
      );
      return row?.head ?? null;
    },

    async putSpaceWriter(space, did, repoRev, data, spaceRev) {
      stmts.putWriter.run(space, did, repoRev, data, spaceRev);
    },

    async removeSpaceWriter(space, did) {
      stmts.removeWriter.run(space, did);
    },

    async listSpaceWriters(space, cursor, limit) {
      const rows = /** @type {SpaceWriterDbRow[]} */ (
        stmts.listWriters.all(space, cursor ?? '', limit)
      );
      const writers = rows.map(toWriterRow);
      return {
        cursor: writers.length ? writers[writers.length - 1].spaceRev : null,
        writers,
      };
    },

    async putCredentialRevocations(space, jtis, expiresAt) {
      stmts.deleteExpiredRevocations.run(new Date().toISOString());
      for (const jti of jtis) stmts.putRevocation.run(space, jti, expiresAt);
    },

    async isCredentialRevoked(space, jti) {
      return Boolean(
        stmts.getRevocation.get(space, jti, new Date().toISOString()),
      );
    },

    async enqueueSpaceNotify(row) {
      stmts.enqueueNotify.run(
        row.space,
        row.repoRev,
        row.data,
        row.nextAttemptAt,
        row.deadlineAt,
      );
    },

    async listDueSpaceNotify(now, limit) {
      const rows = /** @type {SpaceNotifyDbRow[]} */ (
        stmts.listDueNotify.all(now, limit)
      );
      return rows.map(toNotifyRow);
    },

    async rescheduleSpaceNotify(space, repoRev, attempts, nextAttemptAt) {
      stmts.rescheduleNotify.run(attempts, nextAttemptAt, space, repoRev);
    },

    async deleteSpaceNotify(space, repoRev) {
      stmts.deleteNotify.run(space, repoRev);
    },

    async enqueueSpaceDeletedNotify(row) {
      stmts.enqueueDeletedNotify.run(
        row.space,
        row.service,
        row.serviceEndpoint,
        row.attempts,
        row.nextAttemptAt,
        row.deadlineAt,
      );
    },

    async listDueSpaceDeletedNotify(now, limit) {
      const rows = /** @type {SpaceDeletedNotifyDbRow[]} */ (
        stmts.listDueDeletedNotify.all(now, limit)
      );
      return rows.map(toDeletedNotifyRow);
    },

    async rescheduleSpaceDeletedNotify(
      space,
      service,
      attempts,
      nextAttemptAt,
    ) {
      stmts.rescheduleDeletedNotify.run(
        attempts,
        nextAttemptAt,
        space,
        service,
      );
    },

    async deleteSpaceDeletedNotify(space, service) {
      stmts.deleteDeletedNotify.run(space, service);
    },

    async nextSpaceNotifyAt() {
      const row = /** @type {{next: string|null}|undefined} */ (
        stmts.nextNotifyAt.get()
      );
      return row?.next ?? null;
    },

    async putCredentialRecipient(
      space,
      serviceDid,
      serviceEndpoint,
      expiresAt,
    ) {
      stmts.putRecipient.run(space, serviceDid, serviceEndpoint, expiresAt);
    },

    async deleteCredentialRecipient(space, serviceDid) {
      stmts.deleteRecipient.run(space, serviceDid);
    },

    async listCredentialRecipients(space) {
      const rows = /** @type {SpaceRecipientDbRow[]} */ (
        stmts.listRecipients.all(space, new Date().toISOString())
      );
      return rows.map((r) => ({
        serviceDid: r.service_did,
        serviceEndpoint: r.service_endpoint,
        expiresAt: r.expires_at,
      }));
    },
  };
}