// Stand-in for an email provider. Own process, own SQLite file.
// Records every request and, separately, every effect it accepts.
import { createServer } from 'node:http'
import { DatabaseSync } from 'node:sqlite'

const [port, dbPath] = process.argv.slice(2)
const db = new DatabaseSync(dbPath)
db.exec(`
  PRAGMA journal_mode = WAL;
  CREATE TABLE IF NOT EXISTS requests (
    seq INTEGER PRIMARY KEY AUTOINCREMENT, at INTEGER NOT NULL, worker TEXT, attempt_id TEXT,
    operation_id TEXT, idempotency_key TEXT, outcome TEXT NOT NULL, effect_id INTEGER);
  CREATE TABLE IF NOT EXISTS effects (
    effect_id INTEGER PRIMARY KEY AUTOINCREMENT, at INTEGER NOT NULL, operation_id TEXT,
    attempt_id TEXT, worker TEXT, idempotency_key TEXT UNIQUE);
`)
const insertEffect = db.prepare(
  'INSERT OR IGNORE INTO effects (at, operation_id, attempt_id, worker, idempotency_key) VALUES (?, ?, ?, ?, ?)')
const effectByKey = db.prepare('SELECT effect_id FROM effects WHERE idempotency_key = ?')
const insertRequest = db.prepare(
  'INSERT INTO requests (at, worker, attempt_id, operation_id, idempotency_key, outcome, effect_id) VALUES (?, ?, ?, ?, ?, ?, ?)')

function send(body, key) {
  const at = Date.now()
  // One transaction: the dedupe decision and both records commit together.
  db.exec('BEGIN IMMEDIATE')
  try {
    const result = insertEffect.run(at, body.operationId, body.attemptId, body.worker, key ?? null)
    const duplicate = result.changes === 0
    const effectId = duplicate ? effectByKey.get(key).effect_id : Number(result.lastInsertRowid)
    insertRequest.run(at, body.worker, body.attemptId, body.operationId, key ?? null,
      duplicate ? 'duplicate' : 'accepted', effectId)
    db.exec('COMMIT')
    return { accepted: !duplicate, duplicate, effectId }
  } catch (error) {
    db.exec('ROLLBACK')
    throw error
  }
}

createServer((req, res) => {
  let raw = ''
  req.on('data', chunk => { raw += chunk })
  req.on('end', () => {
    try {
      if (req.method === 'POST' && req.url === '/send') {
        const reply = send(JSON.parse(raw), req.headers['idempotency-key'])
        res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify(reply))
      } else if (req.method === 'GET' && req.url === '/state') {
        res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify({
          requests: db.prepare('SELECT * FROM requests ORDER BY seq').all(),
          effects: db.prepare('SELECT * FROM effects ORDER BY effect_id').all(),
        }))
      } else res.writeHead(404).end()
    } catch (error) {
      res.writeHead(500).end(String(error))
    }
  })
}).listen(Number(port), '127.0.0.1', () => process.send?.({ event: 'receiver.ready' }))