import postgres, { type Sql } from 'postgres';
import type { InboxStore } from '@shade/inbox-server';
import { toBase64, fromBase64 } from '@shade/core';
import { ensureInboxServerTables } from './ensure-tables.js';

/**
 * PostgreSQL-backed InboxStore for the Shade Inbox Server (V3.6).
 *
 * Concurrent-safe: insertions are unique on (address, msg_id), so two
 * simultaneous PUTs of the same blob fold into one row via
 * ON CONFLICT. `received_at` is generated server-side from a sequence
 * — strictly monotonic across processes — so cursor pagination remains
 * total-ordered even on multi-instance deployments.
 */
export class PostgresInboxStore implements InboxStore {
  private constructor(
    private readonly sql: Sql,
    private readonly ownsConnection: boolean,
  ) {}

  static async create(connectionString: string): Promise<PostgresInboxStore> {
    const sql = postgres(connectionString);
    const store = new PostgresInboxStore(sql, true);
    await ensureInboxServerTables(sql);
    return store;
  }

  static async fromClient(sql: Sql): Promise<PostgresInboxStore> {
    const store = new PostgresInboxStore(sql, false);
    await ensureInboxServerTables(sql);
    return store;
  }

  async close(): Promise<void> {
    if (this.ownsConnection) await this.sql.end();
  }

  async saveAddressOwner(address: string, signingKey: Uint8Array): Promise<void> {
    await this.sql`
      INSERT INTO shade_inbox_owners (address, signing_key)
      VALUES (${address}, ${toBase64(signingKey)})
      ON CONFLICT (address) DO UPDATE SET signing_key = EXCLUDED.signing_key
    `;
  }

  async getAddressOwner(address: string): Promise<Uint8Array | null> {
    const rows = await this.sql<Array<{ signing_key: string }>>`
      SELECT signing_key FROM shade_inbox_owners WHERE address = ${address}
    `;
    if (rows.length === 0) return null;
    return fromBase64(rows[0]!.signing_key);
  }

  async putBlob(args: {
    address: string;
    msgId: string;
    ciphertext: Uint8Array;
    expiresAt: number;
    senderFp?: string;
  }): Promise<{ created: boolean; receivedAt: number }> {
    // ON CONFLICT DO NOTHING + RETURNING keeps it idempotent and atomic.
    // When a row already exists, we look up its received_at in a follow-up
    // SELECT.
    const inserted = await this.sql<Array<{ received_at: string }>>`
      INSERT INTO shade_inbox_blobs (address, msg_id, ciphertext, received_at, expires_at, sender_fp)
      VALUES (
        ${args.address},
        ${args.msgId},
        ${toBase64(args.ciphertext)},
        nextval('shade_inbox_seq'),
        ${args.expiresAt},
        ${args.senderFp ?? null}
      )
      ON CONFLICT (address, msg_id) DO NOTHING
      RETURNING received_at::text
    `;
    if (inserted.length > 0) {
      return { created: true, receivedAt: parseInt(inserted[0]!.received_at, 10) };
    }
    const existing = await this.sql<Array<{ received_at: string }>>`
      SELECT received_at::text FROM shade_inbox_blobs
      WHERE address = ${args.address} AND msg_id = ${args.msgId}
    `;
    return {
      created: false,
      receivedAt: parseInt(existing[0]!.received_at, 10),
    };
  }

  async fetchBlobs(args: {
    address: string;
    sinceCursor: number;
    now: number;
    limit: number;
  }): Promise<
    Array<{
      msgId: string;
      ciphertext: Uint8Array;
      receivedAt: number;
      expiresAt: number;
      senderFp?: string;
    }>
  > {
    const rows = await this.sql<Array<{
      msg_id: string;
      ciphertext: string;
      received_at: string;
      expires_at: string;
      sender_fp: string | null;
    }>>`
      SELECT msg_id, ciphertext, received_at::text, expires_at::text, sender_fp
      FROM shade_inbox_blobs
      WHERE address = ${args.address}
        AND received_at > ${args.sinceCursor}
        AND expires_at > ${args.now}
      ORDER BY received_at ASC
      LIMIT ${args.limit}
    `;
    return rows.map((r) => {
      const out: {
        msgId: string;
        ciphertext: Uint8Array;
        receivedAt: number;
        expiresAt: number;
        senderFp?: string;
      } = {
        msgId: r.msg_id,
        ciphertext: fromBase64(r.ciphertext),
        receivedAt: parseInt(r.received_at, 10),
        expiresAt: parseInt(r.expires_at, 10),
      };
      if (r.sender_fp) out.senderFp = r.sender_fp;
      return out;
    });
  }

  async deleteBlob(address: string, msgId: string): Promise<boolean> {
    const result = await this.sql`
      DELETE FROM shade_inbox_blobs WHERE address = ${address} AND msg_id = ${msgId}
    `;
    return result.count > 0;
  }

  async countBlobs(address: string, now: number): Promise<number> {
    const rows = await this.sql<Array<{ count: string }>>`
      SELECT COUNT(*)::text AS count FROM shade_inbox_blobs
      WHERE address = ${address} AND expires_at > ${now}
    `;
    return parseInt(rows[0]!.count, 10);
  }

  async purgeExpired(now: number): Promise<number> {
    const result = await this.sql`
      DELETE FROM shade_inbox_blobs WHERE expires_at <= ${now}
    `;
    return result.count;
  }

  async deleteAddress(address: string): Promise<void> {
    await this.sql.begin(async (tx) => {
      await tx`DELETE FROM shade_inbox_owners WHERE address = ${address}`;
      await tx`DELETE FROM shade_inbox_blobs WHERE address = ${address}`;
    });
  }
}
