// One Mastra process. Workers share a libSQL file; nothing else connects them.
import { appendFileSync } from 'node:fs'
import { randomUUID } from 'node:crypto'
import { Mastra } from '@mastra/core'
import { createTool } from '@mastra/core/tools'
import { LibSQLStore } from '@mastra/libsql'
import { z } from 'zod'

const { WORKER, DB_URL, RECEIVER, EVENTS, LEASE_MS } = process.env
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))

function log(event, data = {}) {
  const entry = { at: Date.now(), worker: WORKER, pid: process.pid, event, ...data }
  appendFileSync(EVENTS, JSON.stringify(entry) + '\n')
  process.send?.(entry)
}

const sendReceipt = createTool({
  id: 'send_receipt',
  description: 'Send a payment receipt email (synthetic destination).',
  inputSchema: z.object({
    operationId: z.string(),
    useIdempotencyKey: z.boolean(),
    preSendMs: z.number(),
    postEffectMs: z.number(),
  }),
  execute: async (input, context) => {
    const { operationId, useIdempotencyKey, preSendMs, postEffectMs } = input
    const attemptId = `${WORKER}-${randomUUID().slice(0, 8)}`
    const signal = context?.abortSignal
    signal?.addEventListener('abort', () => log('attempt.aborted', { attemptId, reason: String(signal.reason?.message ?? signal.reason) }), { once: true })
    log('attempt.start', { attemptId, operationId })
    await sleep(preSendMs)
    log('send.request', { attemptId, operationId, idempotencyKey: useIdempotencyKey ? operationId : null })
    const response = await fetch(`${RECEIVER}/send`, {
      method: 'POST',
      headers: { 'content-type': 'application/json', ...(useIdempotencyKey ? { 'idempotency-key': operationId } : {}) },
      body: JSON.stringify({ operationId, attemptId, worker: WORKER }),
    })
    const reply = await response.json()
    log('send.response', { attemptId, operationId, ...reply })
    await sleep(postEffectMs) // follow-up work after the destination accepted
    log('attempt.finish', { attemptId, operationId, effectId: reply.effectId })
    return { attemptId, worker: WORKER, effectId: reply.effectId, duplicate: reply.duplicate }
  },
})

const mastra = new Mastra({
  storage: new LibSQLStore({ id: 'storage', url: DB_URL }),
  tools: { send_receipt: sendReceipt },
  backgroundTasks: {
    enabled: true,
    leaseDurationMs: Number(LEASE_MS),
    defaultTimeoutMs: 60_000,
    recoverStaleTasksOnStart: true,
  },
})
const manager = mastra.backgroundTaskManager
await manager.init()
log('worker.ready', { ownerId: manager.ownerId })

process.on('message', async message => {
  if (message.cmd !== 'enqueue') return
  const { task } = await manager.enqueue({
    runId: `run-${message.args.operationId}`,
    toolName: 'send_receipt',
    toolCallId: `call-${message.args.operationId}`,
    agentId: 'receipts-agent',
    args: message.args,
    maxRetries: message.maxRetries,
    timeoutMs: 60_000,
  })
  log('task.enqueued', { taskId: task.id })
})

process.on('SIGTERM', async () => {
  log('worker.shutdown.start')
  await mastra.shutdown()
  log('worker.shutdown.done')
  process.exit(0)
})

setInterval(() => {}, 1 << 30)