// @pdsjs/node - Node.js HTTP server adapter

import { createHash } from 'node:crypto';
import { createServer as createHttpServer } from 'node:http';
import { createConnection } from 'node:net';
import { secp256k1 } from '@noble/curves/secp256k1';
import { createFsBackupTarget, createFsBlobs } from '@pdsjs/blobs-fs';
import { PersonalDataServer } from '@pdsjs/core';
import { LexiconResolver } from '@pdsjs/lexicon-resolver';
import { createActorStorage, createSharedStorage } from '@pdsjs/storage-sqlite';
import { WebSocketServer } from 'ws';

/** How often extension background work is checked for anything due. */
const EXTENSION_TICK_MS = 10 * 1000;

/**
 * node:crypto behind the core Hasher interface, for streamed blob uploads.
 * @returns {import('@pdsjs/core/ports').Hasher}
 */
function createNodeHasher() {
  const hash = createHash('sha256');
  return {
    /** @param {Uint8Array} chunk */
    update(chunk) {
      hash.update(chunk);
    },
    digest() {
      return new Uint8Array(hash.digest());
    },
  };
}

/**
 * Verify a secp256k1 signature the way the reference stack does: SHA-256 the
 * data, then require a compact, low-S signature. Malformed input returns false
 * rather than throwing. This is the injected half of the signature verifier —
 * P-256 lives in @pdsjs/core/verify; secp256k1 needs @noble/curves, which core
 * deliberately does not depend on.
 *
 * @param {Uint8Array} publicKey - compressed 33-byte point from a did:key
 * @param {Uint8Array} data - the signed bytes
 * @param {Uint8Array} sig - 64-byte compact r||s
 * @returns {Promise<boolean>}
 */
export async function secp256k1Verify(publicKey, data, sig) {
  try {
    const hash = new Uint8Array(
      await crypto.subtle.digest('SHA-256', /** @type {BufferSource} */ (data)),
    );
    return secp256k1.verify(sig, hash, publicKey, {
      format: 'compact',
      lowS: true,
    });
  } catch {
    return false;
  }
}

/**
 * Sign with secp256k1 account keys — the curve the reference PDS gives every
 * account, and the one some third-party services (Go feed generators) are
 * alone able to verify. WebCrypto cannot sign this curve, so core takes the
 * implementation as a port, mirroring the verifier above.
 *
 * @type {import('@pdsjs/core/ports').Secp256k1SignerPort}
 */
export const k256Signer = {
  generatePrivateKey: () => secp256k1.utils.randomPrivateKey(),
  getPublicKey: (privateKey, compressed = true) =>
    secp256k1.getPublicKey(privateKey, compressed),
  async sign(privateKey, data) {
    const hash = new Uint8Array(
      await crypto.subtle.digest('SHA-256', /** @type {BufferSource} */ (data)),
    );
    // Low-S, compact r||s: what atproto verifies and what secp256k1Verify
    // above requires.
    return secp256k1.sign(hash, privateKey, { lowS: true }).toCompactRawBytes();
  },
};

/**
 * @typedef {Object} PdsServer
 * @property {PersonalDataServer} pds - The PDS instance
 * @property {import('node:http').Server} server - HTTP server
 * @property {Object} db - SQLite database instance
 * @property {() => Promise<import('node:http').Server>} listen - Start server
 * @property {() => Promise<void>} close - Close server
 */

/**
 * Create WebSocket port for Node.js
 * @param {WeakMap<Request, import('ws').WebSocket>} upgradeMap
 * @param {import('ws').WebSocketServer} wss - WebSocket server for broadcast
 * @returns {import('@pdsjs/core/ports').WebSocketPort}
 */
function createWebSocket(upgradeMap, wss) {
  return {
    isUpgrade(request) {
      return upgradeMap.has(request);
    },

    upgrade(request, onConnect) {
      const ws = upgradeMap.get(request);
      if (!ws) {
        return new Response('WebSocket upgrade failed', { status: 500 });
      }

      onConnect({
        send(data) {
          if (ws.readyState === ws.OPEN) {
            ws.send(data);
          }
        },
        close() {
          ws.close();
        },
      });

      // Return 200 - actual upgrade handled by server's 'upgrade' event
      // Node.js Response doesn't support status 101
      return new Response(null, {
        status: 200,
        headers: { 'X-WebSocket-Upgraded': 'true' },
      });
    },

    broadcast(data) {
      // Send to all connected WebSocket clients
      for (const client of wss.clients) {
        if (client.readyState === client.OPEN) {
          try {
            client.send(data);
          } catch {
            // Client may have disconnected
          }
        }
      }
    },
  };
}

/**
 * A mail sender that writes the message to stderr instead of delivering it.
 *
 * For local development: it makes the verification code visible without a
 * mail provider. Never point a deployment at it — every code it "sends"
 * lands in the server log, where anyone with log access can read it.
 * @returns {import('@pdsjs/core/ports').EmailPort}
 */
export function createLogEmailer() {
  return {
    async send({ to, subject, text }) {
      console.error(
        `\n--- email (not sent) ---\nTo: ${to}\nSubject: ${subject}\n\n${text}\n------------------------\n`,
      );
    },
  };
}

/**
 * A mail sender that delivers over SMTP, for pointing local development at a
 * catch-all inbox such as Mailpit so sent mail is visible in a browser. It
 * speaks only enough of the protocol for an unauthenticated, non-TLS relay,
 * which is what those dev inboxes expose. Never point it at a real provider.
 * @param {Object} [options]
 * @param {string} [options.host='127.0.0.1'] - SMTP host
 * @param {number} [options.port=1025] - SMTP port
 * @param {string} [options.from='no-reply@localhost'] - Envelope and From address
 * @returns {import('@pdsjs/core/ports').EmailPort}
 */
export function createSmtpEmailer({
  host = '127.0.0.1',
  port = 1025,
  from = 'no-reply@localhost',
} = {}) {
  return {
    send({ to, subject, text, html }) {
      return deliverOverSmtp({ host, port, from, to, subject, text, html });
    },
  };
}

/**
 * Build an RFC 5322 message, with CRLF line endings and leading dots doubled
 * so a body line of "." cannot end the DATA stage early. With `html` present
 * the two bodies are carried as multipart/alternative, so a client shows the
 * HTML and falls back to the text.
 * @param {{from: string, to: string, subject: string, text: string, html?: string}} message
 */
function buildMessage({ from, to, subject, text, html }) {
  const headers = [
    `From: ${from}`,
    `To: ${to}`,
    `Subject: ${subject}`,
    `Date: ${new Date().toUTCString()}`,
    'MIME-Version: 1.0',
  ];
  let body;
  if (html) {
    const boundary = `=_pdsjs_${crypto.randomUUID()}`;
    headers.push(`Content-Type: multipart/alternative; boundary="${boundary}"`);
    body = [
      '',
      `--${boundary}`,
      'Content-Type: text/plain; charset=utf-8',
      '',
      ...text.split(/\r?\n/),
      `--${boundary}`,
      'Content-Type: text/html; charset=utf-8',
      '',
      ...html.split(/\r?\n/),
      `--${boundary}--`,
    ];
  } else {
    headers.push('Content-Type: text/plain; charset=utf-8', '');
    body = text.split(/\r?\n/);
  }
  return [...headers, ...body]
    .map((line) => (line.startsWith('.') ? `.${line}` : line))
    .join('\r\n');
}

/**
 * @param {{host: string, port: number, from: string, to: string, subject: string, text: string, html?: string}} options
 * @returns {Promise<void>}
 */
function deliverOverSmtp({ host, port, from, to, subject, text, html }) {
  return new Promise((resolve, reject) => {
    const socket = createConnection({ host, port });
    socket.setEncoding('utf8');
    socket.setTimeout(10000);

    // Each step waits for a reply carrying `expect`, then sends the next line.
    // The opening step has no line to send: it just awaits the 220 greeting.
    const script = [
      { expect: 220 },
      { send: 'EHLO localhost', expect: 250 },
      { send: `MAIL FROM:<${from}>`, expect: 250 },
      { send: `RCPT TO:<${to}>`, expect: 250 },
      { send: 'DATA', expect: 354 },
      {
        send: `${buildMessage({ from, to, subject, text, html })}\r\n.`,
        expect: 250,
      },
      { send: 'QUIT', expect: 221 },
    ];

    let step = 0;
    let buffer = '';
    const fail = (/** @type {unknown} */ error) => {
      socket.destroy();
      reject(error instanceof Error ? error : new Error(String(error)));
    };

    socket.on('error', fail);
    socket.on('timeout', () => fail(new Error('SMTP connection timed out')));
    socket.on('data', (/** @type {string} */ chunk) => {
      buffer += chunk;
      // A reply may span several lines (`250-...` then `250 ...`); the line
      // with a space after the code is the last one.
      let match = buffer.match(/^(?:\d{3}-[^\n]*\n)*(\d{3}) [^\n]*\n/);
      while (match) {
        const code = Number(match[1]);
        buffer = buffer.slice(match[0].length);
        if (code !== script[step].expect)
          return fail(
            new Error(`SMTP replied ${code}, expected ${script[step].expect}`),
          );
        step += 1;
        if (step >= script.length) {
          socket.end();
          return resolve(undefined);
        }
        socket.write(`${script[step].send}\r\n`);
        match = buffer.match(/^(?:\d{3}-[^\n]*\n)*(\d{3}) [^\n]*\n/);
      }
    });
  });
}

/**
 * Create Node.js PDS server
 * @param {Object} options
 * @param {string} options.dbPath - Path to SQLite database
 * @param {string} [options.blobsDir] - Directory for blob storage (used if blobs not provided)
 * @param {import('@pdsjs/core/ports').BlobPort} [options.blobs] - Custom blob storage adapter
 * @param {string} options.jwtSecret - JWT signing secret
 * @param {number} [options.blobUploadLimit] - Max blob upload size in bytes (default 50 MiB). Bluesky video posting needs room for the processed video the video service writes back, up to its 300MB cap.
 * @param {number} [options.port=3000] - Server port
 * @param {string} [options.hostname] - PDS hostname
 * @param {string} [options.appviewUrl] - AppView URL for proxying
 * @param {string} [options.appviewDid] - AppView DID for service auth
 * @param {string} [options.relayUrl] - Relay URL for firehose notifications (e.g., http://localhost:2470)
 * @param {string} [options.plcUrl] - PLC directory URL for identity operations
 * @param {string} [options.password] - Password for createSession
 * @param {import('@pdsjs/core/ports').LexiconResolverPort} [options.lexiconResolver] - Lexicon resolver for record validation
 * @param {import('@pdsjs/core/ports').EmailPort} [options.emailer] - Sends account mail; without one an address cannot be confirmed
 * @param {boolean} [options.accountUi] - Serve the account application at /account. Needs the account-ui package (an optional peer dependency); without it the server-rendered pages are served instead.
 * @param {string} [options.backupDir] - Directory scheduled backups land in, ideally on a different disk than the database and blobs
 * @param {import('@pdsjs/core/ports').BackupTargetPort} [options.backupTarget] - Custom backup target (createS3BackupTarget from the blobs-s3 package, say); takes precedence over backupDir. Neither given, the backups page reports backups as unavailable.
 * @param {boolean} [options.spaces] - Enable permissioned data (proposal 0016).
 *   Requires the spaces package to be installed; it is an optional peer dependency.
 * @param {{git?: {http?: boolean, browse?: boolean, external?: boolean, guardOnly?: boolean}, sites?: {domain?: string, apexName?: string, queries?: boolean}, drive?: boolean, oci?: boolean, npm?: boolean}} [options.experimental] - Feature
 *   experiments, exempt from semver: anything in this group may change shape
 *   or disappear in any release, including patches. A feature graduates by
 *   moving to a top-level option.
 *
 *   `git` (requires the git package, an optional peer dependency): `http`
 *   serves read-only git smart HTTP under /git/ for dev.pdsjs.git.repo
 *   records (default off); `browse` backs the account page's file tree
 *   (default on when the package is installed). `external: true` turns the
 *   embedded hosting off so a separate git appview serves the reads while
 *   this PDS stays the data store; `guardOnly: true` keeps the one piece
 *   that cannot move, the branch-protection write guard, and drops the
 *   routes. `external` implies `guardOnly`.
 *
 *   `sites` (requires the sites package, an optional peer dependency):
 *   `domain` is the apex static sites serve under (dev.pdsjs.site.deploy
 *   records + blobs) and setting it enables serving; the reverse proxy must
 *   route the apex and *.domain here, with a wildcard certificate (DNS-01).
 *   `apexName` is the site served at the apex (default "home"). `queries`
 *   serves dev.pdsjs.query.run over this repo's dev.pdsjs.query.def records.
 *
 *   `drive` backs the account page's Drive section over dev.pdsjs.drive.file
 *   records (default on when the drive package, an optional peer dependency,
 *   is installed).
 *
 *   `oci` (requires the oci package, an optional peer dependency) serves an
 *   OCI image registry under /v2/ for dev.pdsjs.oci.repo records (default
 *   off). Pulls are anonymous; pushes authenticate with Basic credentials
 *   createSession accepts.
 *
 *   `npm` (requires the npm package, an optional peer dependency) serves an
 *   npm registry under /npm/ for dev.pdsjs.npm.package records (default
 *   off). Installs are anonymous; publishes authenticate with credentials
 *   createSession accepts, stored by npm login as the auth token.
 * @returns {Promise<PdsServer>}
 */
export async function createServer({
  dbPath,
  blobsDir,
  blobs: blobsArg,
  jwtSecret,
  blobUploadLimit,
  port = 3000,
  hostname,
  appviewUrl,
  appviewDid,
  relayUrl,
  plcUrl,
  password,
  lexiconResolver,
  emailer,
  accountUi = false,
  spaces = false,
  experimental = {},
  backupDir,
  backupTarget: backupTargetArg,
}) {
  const {
    git = {},
    sites = {},
    drive = true,
    oci = false,
    npm = false,
  } = experimental;
  const gitHttp = Boolean(git.http);
  const gitBrowse = git.browse !== false;
  const {
    domain: siteDomain,
    apexName: siteApexName,
    queries: enableQueries,
  } = sites;
  // Dynamic import for better-sqlite3 (optional peer dependency)
  const Database = (await import('better-sqlite3')).default;

  const db = new Database(dbPath);

  // Create actor storage (per-DID data)
  const actorStorage = createActorStorage(db);

  // Create shared storage (global data) - can use same db for single-tenant
  const sharedStorage = createSharedStorage(db);

  // Signature verification, shared by createAccount (migration service tokens)
  // and, when enabled, spaces (credentials). P-256 is built into core;
  // secp256k1 — the curve the reference PDS gives every account — is supplied
  // here via @noble/curves. The DID resolver returns full documents and honours
  // this server's PLC directory rather than the public default.
  const { createVerifier } = await import('@pdsjs/core/verify');
  const { createCachedDidResolver } = await import('@pdsjs/lexicon-resolver');
  const verifier = createVerifier({ secp256k1: secp256k1Verify });
  // One resolver for the whole server, built once so its cache spans requests.
  // Every space request verifies a credential, which reads its issuer's key.
  const didResolver = createCachedDidResolver({ plcUrl });

  if (!blobsArg && !blobsDir) {
    throw new Error('Either blobs or blobsDir must be provided');
  }
  const blobs = blobsArg ?? createFsBlobs(/** @type {string} */ (blobsDir));

  /** @type {import('@pdsjs/core/ports').SpaceStoragePort|undefined} */
  let spaceStorageRef;

  // Feature packages on the extension seam, each an optional peer imported
  // only when its option is on. Git is on whenever the package is installed,
  // unless `git.external` hands the reads to a separate git appview: then
  // only the branch-protection write guard stays (guardOnly), and with
  // neither option the full extension serves the routes itself.
  /** @type {import('@pdsjs/core/ports').ExtensionFactory[]} */
  const extensions = [];
  // Static sites and apps: the installs listing needs no option, a site
  // domain turns serving on.
  try {
    const { createSitesExtension } = await import('@pdsjs/sites/extension');
    extensions.push(
      createSitesExtension({
        domain: siteDomain,
        apexName: siteApexName,
        queries: enableQueries,
      }),
    );
  } catch {
    // Optional peer not installed
  }
  try {
    if (git.external || git.guardOnly) {
      // The reads are served elsewhere; only the write-time enforcement
      // stays, and the routes drop with it.
      const { createGitGuardExtension } = await import('@pdsjs/git/extension');
      extensions.push(createGitGuardExtension());
    } else {
      const { createGitExtension } = await import('@pdsjs/git/extension');
      extensions.push(createGitExtension({ http: gitHttp, browse: gitBrowse }));
    }
  } catch {
    // Optional peer not installed
  }

  // Permissioned data (proposal 0016), off by default. The storage is this
  // platform's; the space endpoints, the admin port and the account page's
  // space routes come from @pdsjs/spaces, an optional peer imported only
  // here. With this off the space endpoints do not exist and requests fall
  // through to the normal 404, which matters because space-host discovery
  // falls back to #atproto_pds: a space-unaware PDS still receives space
  // requests from optimistic clients.
  if (spaces) {
    const [{ createSpaceStorage }, { createSpacesExtension }] =
      await Promise.all([
        import('@pdsjs/storage-sqlite'),
        import('@pdsjs/spaces/extension'),
      ]);
    spaceStorageRef = createSpaceStorage(db);
    extensions.push(createSpacesExtension());
  }

  // The built application arrives as a module rather than a directory, so it
  // loads the same way on every platform. Optional: absent, /account keeps
  // serving the server-rendered pages.
  let accountApp;
  if (accountUi) {
    ({ accountApp } = await import('@pdsjs/account-ui'));
  }

  const backupTarget =
    backupTargetArg ??
    (backupDir ? createFsBackupTarget(backupDir) : undefined);

  // WebSocket infrastructure
  const upgradeMap = new WeakMap();
  const wss = new WebSocketServer({ noServer: true });
  const webSocket = createWebSocket(upgradeMap, wss);

  // Create default lexicon resolver if not provided
  const resolver = lexiconResolver ?? new LexiconResolver();

  // Files stored in the repo, for the account page's Drive section: on
  // whenever the drive package (an optional peer) is installed, unless
  // refused.
  if (drive) {
    try {
      const { createDriveExtension } = await import('@pdsjs/drive/extension');
      extensions.push(createDriveExtension());
    } catch {
      // Optional peer not installed
    }
  }

  if (oci) {
    const { createOciExtension } = await import('@pdsjs/oci/extension');
    extensions.push(createOciExtension());
  }
  if (npm) {
    const { createNpmExtension } = await import('@pdsjs/npm/extension');
    extensions.push(createNpmExtension());
  }

  // Create PDS with both storages
  const pds = new PersonalDataServer({
    actorStorage,
    sharedStorage,
    blobs,
    webSocket,
    jwtSecret,
    blobUploadLimit,
    hasher: createNodeHasher,
    hostname,
    appviewUrl,
    appviewDid,
    relayUrl,
    plcUrl,
    password,
    lexiconResolver: resolver,
    emailer,
    spaceStorage: spaceStorageRef,
    accountApp,
    verifier,
    didResolver,
    k256Signer,
    backupTarget,
    extensions,
  });

  const server = createHttpServer(async (req, res) => {
    try {
      const request = nodeToWebRequest(req);
      const response = await pds.fetch(request);

      // Skip normal response for WebSocket upgrades
      if (
        response.headers.get('X-WebSocket-Upgraded') &&
        upgradeMap.has(request)
      ) {
        return; // WebSocket handled by upgrade event
      }

      await webToNodeResponse(response, res);
    } catch (error) {
      console.error('Server error:', error);
      res.statusCode = 500;
      res.setHeader('Content-Type', 'application/json');
      res.end(
        JSON.stringify({
          error: 'InternalServerError',
          message: error.message,
        }),
      );
    }
  });

  // Handle WebSocket upgrades
  server.on('upgrade', (req, socket, head) => {
    if (req.url?.startsWith('/xrpc/com.atproto.sync.subscribeRepos')) {
      wss.handleUpgrade(req, socket, head, async (ws) => {
        const request = nodeToWebRequest(req);
        upgradeMap.set(request, ws);
        try {
          await pds.fetch(request);
        } finally {
          upgradeMap.delete(request);
        }
      });
    } else {
      socket.destroy();
    }
  });

  // Start maintenance interval (every hour): cleanup, then a backup if due
  const cleanupInterval = setInterval(
    async () => {
      try {
        await pds.runBlobCleanup();
        await sharedStorage.cleanupExpiredDpopJtis();
        await pds.maybeAnnounceIdentity();
        await pds.maybeBackfillKemKey();
        await pds.maybeRunScheduledBackup();
      } catch (e) {
        console.error('Maintenance error:', e);
      }
    },
    60 * 60 * 1000,
  );

  // An account created before KEM keys gets one now rather than in an hour.
  pds.maybeBackfillKemKey();

  // Extension background work, such as retrying space write notices. A tick
  // with nothing due is one indexed query.
  const tickInterval = setInterval(() => {
    pds.runExtensionTicks().catch((e) => console.error('Tick error:', e));
  }, EXTENSION_TICK_MS);

  return {
    pds,
    server,
    db,

    /**
     * Start listening on configured port
     * @returns {Promise<import('node:http').Server>}
     */
    listen() {
      return new Promise((resolve) => {
        server.listen(port, () => {
          console.log(`PDS listening on http://localhost:${port}`);
          resolve(server);
        });
      });
    },

    /**
     * Close server and cleanup
     * @returns {Promise<void>}
     */
    close() {
      clearInterval(cleanupInterval);
      clearInterval(tickInterval);
      // Close all WebSocket connections first
      wss.clients.forEach((client) => {
        client.terminate();
      });
      wss.close();
      db.close();
      return new Promise((resolve, reject) => {
        server.close((err) => {
          if (err) reject(err);
          else resolve();
        });
      });
    },
  };
}

/**
 * Convert Node.js IncomingMessage to Web Request
 * @param {import('node:http').IncomingMessage} req
 * @returns {Request}
 */
function nodeToWebRequest(req) {
  // Check for encrypted socket (HTTPS)
  const socket = /** @type {import('node:tls').TLSSocket} */ (req.socket);
  const protocol = socket.encrypted ? 'https' : 'http';
  const host = req.headers.host || 'localhost';
  const url = new URL(req.url || '/', `${protocol}://${host}`);

  const headers = new Headers();
  for (const [key, value] of Object.entries(req.headers)) {
    if (value) {
      headers.set(key, Array.isArray(value) ? value.join(', ') : value);
    }
  }

  const hasBody = req.method !== 'GET' && req.method !== 'HEAD';

  return new Request(
    url.toString(),
    /** @type {RequestInit} */ ({
      method: req.method,
      headers,
      body: hasBody ? req : null,
      duplex: hasBody ? 'half' : undefined,
    }),
  );
}

/**
 * Write Web Response to Node.js ServerResponse
 * @param {Response} response
 * @param {import('node:http').ServerResponse} res
 */
async function webToNodeResponse(response, res) {
  res.statusCode = response.status;
  res.statusMessage = response.statusText || '';

  for (const [key, value] of response.headers) {
    res.setHeader(key, value);
  }

  if (response.body) {
    const reader = response.body.getReader();
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      // A false return means the socket's buffer is full. Waiting for drain
      // holds streamed blobs to one chunk in memory instead of the whole
      // body.
      if (!res.write(value)) {
        await new Promise((resolve) => res.once('drain', resolve));
      }
    }
  }

  res.end();
}