// @pdsjs/cloudflare/space - Durable Object SQLite SpaceStoragePort (proposal 0016).
//
// Shares its schema with the storage-sqlite adapter via @pdsjs/core/space-schema;
// only the driver differs. Behaviour is pinned by the shared conformance suite
// in test/helpers/space-storage-conformance.js, which both adapters run.

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: ArrayBuffer|Uint8Array, repo_rev: string, indexed_at: string}} SpaceRecordDbRow */
/** @typedef {Omit<SpaceRecordDbRow, 'value'> & {value?: ArrayBuffer|Uint8Array}} 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: ArrayBuffer|Uint8Array, space_rev: string}} SpaceWriterDbRow */
/** @typedef {{space: string, repo_rev: string, data: ArrayBuffer|Uint8Array, 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 */

/**
 * Durable Object SQLite returns ArrayBuffer for BLOB columns.
 * @param {ArrayBuffer|Uint8Array|null|undefined} value
 * @returns {Uint8Array|null}
 */
function normalizeBlob(value) {
  if (!value) return null;
  if (value instanceof Uint8Array) return new Uint8Array(value);
  return 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} */ (normalizeBlob(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} */ (normalizeBlob(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} */ (normalizeBlob(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} */ (normalizeBlob(row.value)) }),
    repoRev: row.repo_rev,
    indexedAt: row.indexed_at,
  };
}

/**
 * @param {any} sql - D1SqlInterface, from state.storage.sql
 * @param {{transactionSync: (fn: () => any) => any}} storage - state.storage
 * @returns {import('@pdsjs/core/ports').SpaceStoragePort}
 */
export function createSpaceStorage(sql, storage) {
  sql.exec(SPACE_SCHEMA_SQL);
  for (const statement of SPACE_MIGRATIONS) {
    try {
      sql.exec(statement);
    } catch {
      // Already applied.
    }
  }

  /**
   * Single-row read. Do NOT use cursor.one() — it throws on an empty result,
   * which is why the actor-store adapter in index.js carries its own safeOne.
   * @param {string} query
   * @param {...any} params
   * @returns {any|null}
   */
  function one(query, ...params) {
    const rows = sql.exec(query, ...params).toArray();
    return rows.length > 0 ? rows[0] : null;
  }

  /**
   * @param {string} query
   * @param {...any} params
   * @returns {any[]}
   */
  function all(query, ...params) {
    return sql.exec(query, ...params).toArray();
  }

  return {
    async getSpace(uri) {
      const row = one('SELECT * FROM space WHERE uri = ?', uri);
      return row ? toSpaceRow(row) : null;
    },

    async putSpace(space) {
      sql.exec(
        `INSERT OR REPLACE INTO space
         (uri, space_did, space_type, is_owner, policy_json, created_at, deleted_at)
         VALUES (?, ?, ?, ?, ?, ?, ?)`,
        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) {
      sql.exec('UPDATE space SET deleted_at = ? WHERE uri = ?', 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[]} */ (
        all(
          `SELECT * FROM space WHERE ${where.join(' AND ')} ORDER BY uri LIMIT ?`,
          ...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) {
      sql.exec(
        `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`,
        space,
        did,
        access.read ? 1 : 0,
        access.write ? 1 : 0,
      );
    },

    async removeMember(space, did) {
      sql.exec(
        'DELETE FROM space_member WHERE space = ? AND did = ?',
        space,
        did,
      );
    },

    async listMembers(space, cursor, limit) {
      const rows = /** @type {SpaceMemberDbRow[]} */ (
        all(
          'SELECT did, can_read, can_write FROM space_member WHERE space = ? AND did > ? ORDER BY did LIMIT ?',
          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|null} */ (
        one(
          'SELECT can_read, can_write FROM space_member WHERE space = ? AND did = ?',
          space,
          did,
        )
      );
      return row
        ? { read: row.can_read === 1, write: row.can_write === 1 }
        : null;
    },

    async removeAllMembers(space) {
      sql.exec('DELETE FROM space_member WHERE space = ?', space);
    },

    async getSpaceRecord(space, collection, rkey) {
      const row = one(
        'SELECT collection, rkey, cid, value, repo_rev, indexed_at FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
        space,
        collection,
        rkey,
      );
      return row ? toRecordRow(row) : null;
    },

    async getSpaceRecordCid(space, collection, rkey) {
      const row = one(
        'SELECT cid FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
        space,
        collection,
        rkey,
      );
      return row ? row.cid : null;
    },

    async listSpaceRecords(space, opts) {
      const { sql: query, params } = spaceRecordsQuery(space, opts);
      const rows = /** @type {SpaceRecordListDbRow[]} */ (
        all(query, ...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}>} */ (
        all(
          "SELECT collection, rkey, cid FROM space_record WHERE space = ? ORDER BY (collection || '/' || rkey)",
          space,
        )
      );
    },

    async commitSpaceWrite(space, { writes, setHash, rev, indexedAt }) {
      // Validate everything before opening the transaction, so a bad batch
      // cannot leave a partially applied commit behind.
      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}`,
          );
        }
      }

      // Durable Object SQL rejects explicit BEGIN/COMMIT statements;
      // transactionSync is the supported way to make several statements atomic.
      storage.transactionSync(() => {
        writes.forEach((w, idx) => {
          if (w.action === 'delete') {
            sql.exec(
              'DELETE FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
              space,
              w.collection,
              w.rkey,
            );
          } else {
            sql.exec(
              `INSERT OR REPLACE INTO space_record
               (space, collection, rkey, cid, value, repo_rev, indexed_at)
               VALUES (?, ?, ?, ?, ?, ?, ?)`,
              space,
              w.collection,
              w.rkey,
              w.cid,
              w.value,
              rev,
              indexedAt,
            );
          }
          sql.exec(
            `INSERT INTO space_record_oplog
             (space, rev, idx, action, collection, rkey, cid, prev)
             VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
            space,
            rev,
            idx,
            w.action,
            w.collection,
            w.rkey,
            w.cid,
            w.prev,
          );
        });
        sql.exec(
          'INSERT OR REPLACE INTO space_repo (space, set_hash, rev) VALUES (?, ?, ?)',
          space,
          setHash,
          rev,
        );
      });
    },

    async getSpaceRepo(space) {
      const row = one(
        'SELECT set_hash, rev FROM space_repo WHERE space = ?',
        space,
      );
      if (!row) return null;
      return { setHash: normalizeBlob(row.set_hash), rev: row.rev };
    },

    async listSpaceOps(space, since, limit) {
      const rows = /** @type {SpaceOpDbRow[]} */ (
        all(
          'SELECT rev, idx, action, collection, rkey, cid, prev FROM space_record_oplog WHERE space = ? AND rev > ? ORDER BY rev, idx LIMIT ?',
          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|null} */ (
        one(
          'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND did = ?',
          space,
          did,
        )
      );
      return row ? toWriterRow(row) : null;
    },

    async getSpaceRevHead(space) {
      const row = /** @type {{head: string|null}|null} */ (
        one(
          'SELECT MAX(space_rev) AS head FROM space_writer WHERE space = ?',
          space,
        )
      );
      return row?.head ?? null;
    },

    async removeSpaceWriter(space, did) {
      sql.exec(
        'DELETE FROM space_writer WHERE space = ? AND did = ?',
        space,
        did,
      );
    },

    async putSpaceWriter(space, did, repoRev, data, spaceRev) {
      sql.exec(
        'INSERT OR REPLACE INTO space_writer (space, did, rev, data, space_rev) VALUES (?, ?, ?, ?, ?)',
        space,
        did,
        repoRev,
        data,
        spaceRev,
      );
    },

    async listSpaceWriters(space, cursor, limit) {
      const rows = /** @type {SpaceWriterDbRow[]} */ (
        all(
          'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND space_rev > ? ORDER BY space_rev LIMIT ?',
          space,
          cursor ?? '',
          limit,
        )
      );
      const writers = rows.map(toWriterRow);
      return {
        writers,
        cursor: writers.length ? writers[writers.length - 1].spaceRev : null,
      };
    },

    async putCredentialRevocations(space, jtis, expiresAt) {
      sql.exec(
        'DELETE FROM space_credential_revocation WHERE expires_at <= ?',
        new Date().toISOString(),
      );
      for (const jti of jtis) {
        sql.exec(
          `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)`,
          space,
          jti,
          expiresAt,
        );
      }
    },

    async isCredentialRevoked(space, jti) {
      return Boolean(
        one(
          'SELECT 1 AS revoked FROM space_credential_revocation WHERE space = ? AND jti = ? AND expires_at > ?',
          space,
          jti,
          new Date().toISOString(),
        ),
      );
    },

    async enqueueSpaceNotify(row) {
      sql.exec(
        `INSERT OR REPLACE INTO space_notify_outbox
         (space, repo_rev, data, attempts, next_attempt_at, deadline_at)
         VALUES (?, ?, ?, 0, ?, ?)`,
        row.space,
        row.repoRev,
        row.data,
        row.nextAttemptAt,
        row.deadlineAt,
      );
    },

    async listDueSpaceNotify(now, limit) {
      const rows = /** @type {SpaceNotifyDbRow[]} */ (
        all(
          '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 ?',
          now,
          limit,
        )
      );
      return rows.map(toNotifyRow);
    },

    async rescheduleSpaceNotify(space, repoRev, attempts, nextAttemptAt) {
      sql.exec(
        'UPDATE space_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND repo_rev = ?',
        attempts,
        nextAttemptAt,
        space,
        repoRev,
      );
    },

    async deleteSpaceNotify(space, repoRev) {
      sql.exec(
        'DELETE FROM space_notify_outbox WHERE space = ? AND repo_rev = ?',
        space,
        repoRev,
      );
    },

    async enqueueSpaceDeletedNotify(row) {
      sql.exec(
        `INSERT OR REPLACE INTO space_deleted_notify_outbox
         (space, service, service_endpoint, attempts, next_attempt_at, deadline_at)
         VALUES (?, ?, ?, ?, ?, ?)`,
        row.space,
        row.service,
        row.serviceEndpoint,
        row.attempts,
        row.nextAttemptAt,
        row.deadlineAt,
      );
    },

    async listDueSpaceDeletedNotify(now, limit) {
      const rows = /** @type {SpaceDeletedNotifyDbRow[]} */ (
        all(
          '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 ?',
          now,
          limit,
        )
      );
      return rows.map(toDeletedNotifyRow);
    },

    async rescheduleSpaceDeletedNotify(
      space,
      service,
      attempts,
      nextAttemptAt,
    ) {
      sql.exec(
        'UPDATE space_deleted_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND service = ?',
        attempts,
        nextAttemptAt,
        space,
        service,
      );
    },

    async deleteSpaceDeletedNotify(space, service) {
      sql.exec(
        'DELETE FROM space_deleted_notify_outbox WHERE space = ? AND service = ?',
        space,
        service,
      );
    },

    async nextSpaceNotifyAt() {
      const row = /** @type {{next: string|null}|null} */ (
        one(
          `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
           )`,
        )
      );
      return row?.next ?? null;
    },

    async putCredentialRecipient(
      space,
      serviceDid,
      serviceEndpoint,
      expiresAt,
    ) {
      sql.exec(
        `INSERT OR REPLACE INTO space_credential_recipient
         (space, service_did, service_endpoint, expires_at)
         VALUES (?, ?, ?, ?)`,
        space,
        serviceDid,
        serviceEndpoint,
        expiresAt,
      );
    },

    async deleteCredentialRecipient(space, serviceDid) {
      sql.exec(
        'DELETE FROM space_credential_recipient WHERE space = ? AND service_did = ?',
        space,
        serviceDid,
      );
    },

    async listCredentialRecipients(space) {
      const rows = /** @type {SpaceRecipientDbRow[]} */ (
        all(
          'SELECT service_did, service_endpoint, expires_at FROM space_credential_recipient WHERE space = ? AND expires_at > ? ORDER BY service_did',
          space,
          new Date().toISOString(),
        )
      );
      return rows.map((r) => ({
        serviceDid: r.service_did,
        serviceEndpoint: r.service_endpoint,
        expiresAt: r.expires_at,
      }));
    },
  };
}