/**
 * UOK Coordination Store
 *
 * Purpose: provide Redis-like coordination primitives on top of the project
 * SQLite database so UOK can lease work, keep TTL state, append streams, and
 * coordinate background tasks without adding a server process.
 *
 * Consumer: UOK scheduler, /tasks, parallel dispatch, and future background
 * work controllers that need durable coordination inside `.sf/sf.db`.
 */

const DEFAULT_NOW = () => Date.now();

/**
 * Ensure coordination tables exist in the shared SF database.
 *
 * Purpose: let callers opt into queue, stream, and TTL primitives without
 * coupling them to the main SF schema migration cadence during greenfield UOK
 * iteration.
 *
 * Consumer: UokCoordinationStore constructor and direct tests.
 */
export function ensureCoordinationSchema(db) {
	db.exec(`
    CREATE TABLE IF NOT EXISTS uok_kv (
      key TEXT PRIMARY KEY,
      value_json TEXT NOT NULL,
      expires_at INTEGER DEFAULT NULL,
      updated_at INTEGER NOT NULL
    );

    CREATE INDEX IF NOT EXISTS idx_uok_kv_expires ON uok_kv(expires_at);

    CREATE TABLE IF NOT EXISTS uok_stream_entries (
      stream TEXT NOT NULL,
      id INTEGER NOT NULL,
      type TEXT NOT NULL,
      payload_json TEXT NOT NULL,
      ts INTEGER NOT NULL,
      PRIMARY KEY (stream, id)
    );

    CREATE INDEX IF NOT EXISTS idx_uok_stream_entries_ts
      ON uok_stream_entries(stream, ts);

    CREATE TABLE IF NOT EXISTS uok_queue_items (
      queue TEXT NOT NULL,
      id TEXT NOT NULL,
      payload_json TEXT NOT NULL,
      status TEXT NOT NULL DEFAULT 'ready',
      available_at INTEGER NOT NULL,
      lease_owner TEXT DEFAULT NULL,
      lease_until INTEGER DEFAULT NULL,
      attempts INTEGER NOT NULL DEFAULT 0,
      created_at INTEGER NOT NULL,
      updated_at INTEGER NOT NULL,
      PRIMARY KEY (queue, id)
    );

    CREATE INDEX IF NOT EXISTS idx_uok_queue_ready
      ON uok_queue_items(queue, status, available_at, lease_until);
  `);
}

function encode(value) {
	return JSON.stringify(value ?? null);
}

function decode(text) {
	return text ? JSON.parse(text) : null;
}

function normalizeRow(row) {
	if (row == null) return null;
	return Object.getPrototypeOf(row) === null ? { ...row } : row;
}

/**
 * Durable Redis-like coordination primitives backed by SQLite.
 *
 * Purpose: keep UOK coordination local, inspectable, and serverless while
 * preserving the useful parts of Redis: TTL values, append-only streams, and
 * lease-based queues.
 *
 * Consumer: scheduler v2 and future UOK background workers.
 */
export class UokCoordinationStore {
	constructor(db, { now = DEFAULT_NOW } = {}) {
		this.db = db;
		this.now = now;
		ensureCoordinationSchema(db);
	}

	set(key, value, { ttlMs = null } = {}) {
		const now = this.now();
		const expiresAt = ttlMs == null ? null : now + ttlMs;
		this.db
			.prepare(
				`INSERT INTO uok_kv (key, value_json, expires_at, updated_at)
         VALUES (:key, :value, :expiresAt, :now)
         ON CONFLICT(key) DO UPDATE SET
           value_json = excluded.value_json,
           expires_at = excluded.expires_at,
           updated_at = excluded.updated_at`,
			)
			.run({ key, value: encode(value), expiresAt, now });
		return true;
	}

	get(key) {
		const now = this.now();
		const row = normalizeRow(
			this.db
				.prepare(
					`SELECT value_json, expires_at FROM uok_kv
           WHERE key = :key`,
				)
				.get({ key }),
		);
		if (!row) return null;
		if (row.expires_at != null && row.expires_at <= now) {
			this.delete(key);
			return null;
		}
		return decode(row.value_json);
	}

	entries(prefix = "") {
		this.cleanupExpired();
		return this.db
			.prepare(
				`SELECT key, value_json, expires_at, updated_at
         FROM uok_kv
         WHERE key LIKE :prefix
         ORDER BY updated_at DESC, key ASC`,
			)
			.all({ prefix: `${prefix}%` })
			.map((row) => ({
				key: row.key,
				value: decode(row.value_json),
				expiresAt: row.expires_at,
				updatedAt: row.updated_at,
			}));
	}

	delete(key) {
		this.db.prepare("DELETE FROM uok_kv WHERE key = :key").run({ key });
	}

	expire(key, ttlMs) {
		const result = this.db
			.prepare(
				`UPDATE uok_kv
         SET expires_at = :expiresAt, updated_at = :now
         WHERE key = :key`,
			)
			.run({ key, expiresAt: this.now() + ttlMs, now: this.now() });
		return (result?.changes ?? 0) > 0;
	}

	cleanupExpired() {
		const result = this.db
			.prepare(
				`DELETE FROM uok_kv
         WHERE expires_at IS NOT NULL AND expires_at <= :now`,
			)
			.run({ now: this.now() });
		return result?.changes ?? 0;
	}

	xadd(stream, type, payload) {
		const now = this.now();
		const row = normalizeRow(
			this.db
				.prepare(
					`SELECT COALESCE(MAX(id), 0) + 1 AS next_id
           FROM uok_stream_entries
           WHERE stream = :stream`,
				)
				.get({ stream }),
		);
		const id = row?.next_id ?? 1;
		this.db
			.prepare(
				`INSERT INTO uok_stream_entries (stream, id, type, payload_json, ts)
         VALUES (:stream, :id, :type, :payload, :ts)`,
			)
			.run({ stream, id, type, payload: encode(payload), ts: now });
		return id;
	}

	xread(stream, { afterId = 0, limit = 100 } = {}) {
		return this.db
			.prepare(
				`SELECT id, type, payload_json, ts
         FROM uok_stream_entries
         WHERE stream = :stream AND id > :afterId
         ORDER BY id ASC
         LIMIT :limit`,
			)
			.all({ stream, afterId, limit })
			.map((row) => ({
				id: row.id,
				type: row.type,
				payload: decode(row.payload_json),
				ts: row.ts,
			}));
	}

	enqueue(queue, id, payload, { delayMs = 0 } = {}) {
		const now = this.now();
		this.db
			.prepare(
				`INSERT INTO uok_queue_items (
           queue, id, payload_json, status, available_at, created_at, updated_at
         ) VALUES (
           :queue, :id, :payload, 'ready', :availableAt, :now, :now
         )
         ON CONFLICT(queue, id) DO UPDATE SET
           payload_json = excluded.payload_json,
           status = 'ready',
           available_at = excluded.available_at,
           lease_owner = NULL,
           lease_until = NULL,
           updated_at = excluded.updated_at`,
			)
			.run({
				queue,
				id,
				payload: encode(payload),
				availableAt: now + delayMs,
				now,
			});
		return id;
	}

	claim(queue, owner, { leaseMs = 30_000 } = {}) {
		const now = this.now();
		const row = normalizeRow(
			this.db
				.prepare(
					`SELECT id
           FROM uok_queue_items
           WHERE queue = :queue
             AND status = 'ready'
             AND available_at <= :now
             AND (lease_until IS NULL OR lease_until <= :now)
           ORDER BY available_at ASC, created_at ASC, id ASC
           LIMIT 1`,
				)
				.get({ queue, now }),
		);
		if (!row) return null;

		const result = this.db
			.prepare(
				`UPDATE uok_queue_items
         SET status = 'leased',
             lease_owner = :owner,
             lease_until = :leaseUntil,
             attempts = attempts + 1,
             updated_at = :now
         WHERE queue = :queue
           AND id = :id
           AND status = 'ready'
           AND (lease_until IS NULL OR lease_until <= :now)`,
			)
			.run({
				queue,
				id: row.id,
				owner,
				leaseUntil: now + leaseMs,
				now,
			});
		if ((result?.changes ?? 0) !== 1) return null;

		return this.peekQueueItem(queue, row.id);
	}

	ack(queue, id, owner) {
		const result = this.db
			.prepare(
				`UPDATE uok_queue_items
         SET status = 'done', updated_at = :now
         WHERE queue = :queue
           AND id = :id
           AND lease_owner = :owner
           AND status = 'leased'`,
			)
			.run({ queue, id, owner, now: this.now() });
		return (result?.changes ?? 0) === 1;
	}

	release(queue, id, owner, { delayMs = 0 } = {}) {
		const now = this.now();
		const result = this.db
			.prepare(
				`UPDATE uok_queue_items
         SET status = 'ready',
             available_at = :availableAt,
             lease_owner = NULL,
             lease_until = NULL,
             updated_at = :now
         WHERE queue = :queue
           AND id = :id
           AND lease_owner = :owner
           AND status = 'leased'`,
			)
			.run({ queue, id, owner, availableAt: now + delayMs, now });
		return (result?.changes ?? 0) === 1;
	}

	reapExpiredLeases(queue = null) {
		const now = this.now();
		const sql =
			queue == null
				? `UPDATE uok_queue_items
           SET status = 'ready', lease_owner = NULL, lease_until = NULL, updated_at = :now
           WHERE status = 'leased' AND lease_until <= :now`
				: `UPDATE uok_queue_items
           SET status = 'ready', lease_owner = NULL, lease_until = NULL, updated_at = :now
           WHERE queue = :queue AND status = 'leased' AND lease_until <= :now`;
		const params = queue == null ? { now } : { queue, now };
		const result = this.db.prepare(sql).run(params);
		return result?.changes ?? 0;
	}

	peekQueueItem(queue, id) {
		const row = normalizeRow(
			this.db
				.prepare(
					`SELECT *
           FROM uok_queue_items
           WHERE queue = :queue AND id = :id`,
				)
				.get({ queue, id }),
		);
		if (!row) return null;
		return {
			queue: row.queue,
			id: row.id,
			payload: decode(row.payload_json),
			status: row.status,
			availableAt: row.available_at,
			leaseOwner: row.lease_owner,
			leaseUntil: row.lease_until,
			attempts: row.attempts,
			createdAt: row.created_at,
			updatedAt: row.updated_at,
		};
	}
}
