KV, queues & cron

// main.ts
import { getKv } from '@netscript/kv';

const kv = await getKv(); // picks Redis/Garnet under Aspire, else local Deno KV

await kv.set(['users', 'alice'], { name: 'Alice', role: 'admin' });

const entry = await kv.get<{ name: string; role: string }>(['users', 'alice']);
console.log(entry?.value.name); // "Alice"

// Per-key TTL works on every adapter — expire a session after one hour.
await kv.set(['sessions', 'tok_abc'], { userId: 'u1' }, { expireIn: 3_600_000 });

// watch-jobs.ts
import { getKv } from '@netscript/kv';

const kv = await getKv();

// Stream every change under a prefix, including newly created keys.
for await (const event of kv.watchPrefix(['jobs', 'order-processor'])) {
  console.log(`${event.key.join('/')} -> ${event.type}`, event.value);
}

// counter.ts
import { getKv } from '@netscript/kv';

const kv = await getKv();

const entry = await kv.get<number>(['counters', 'visits']);
const result = await kv.atomic(
  [{ key: ['counters', 'visits'], versionstamp: entry?.versionstamp ?? null }],
  [{ type: 'set', key: ['counters', 'visits'], value: (entry?.value ?? 0) + 1 }],
);

if (!result.ok) {
  // A concurrent write landed since the read — retry the cycle.
}

// kv.test.ts — deterministic, process-local store for tests.
import { MemoryKvAdapter } from '@netscript/kv';

const kv = new MemoryKvAdapter();
await kv.set(['test', 'key'], 'value');

// Or pin Redis explicitly via the subpath import (keeps ioredis out of other graphs).
import { RedisKvAdapter } from '@netscript/kv/redis';

const redisKv = new RedisKvAdapter({ url: 'redis://localhost:6379', namespace: 'myapp' });
await redisKv.set(['cache', 'featured'], items, { expireIn: 60_000 });

// enqueue-email.ts
import { createQueue } from '@netscript/queue';

// Auto-detects: RabbitMQ (AMQP) under Aspire, else Redis, else Deno KV.
const emails = createQueue<{ to: string; body: string }>('emails');

await emails.enqueue({
  to: 'user@example.com',
  body: 'Welcome to NetScript.',
});

// Delay availability (e.g. a 5-minute reminder) via EnqueueOptions.
await emails.enqueue({ to: 'user@example.com', body: 'Still there?' }, { delay: 5 * 60 * 1000 });

// consume-email.ts
import { createQueue } from '@netscript/queue';

const emails = createQueue<{ to: string; body: string }>('emails');

// The handler receives the message and a MessageContext (messageId, deliveryCount, ack/nack).
await emails.listen(async (message, context) => {
  await sendEmail(message.to, message.body);
  // Some backends ack/retry natively (queue.nativeRetrial === true).
});

// On shutdown, drain in-flight work and release the backend connection.
await emails.stop();

// parallel-jobs.ts
import { createParallelQueue } from '@netscript/queue';

// Process up to 4 I/O-bound messages concurrently on a single listener.
const jobs = createParallelQueue<{ orderId: string }>('order-jobs', { concurrency: 4 });

await jobs.listen(async (message) => {
  await fulfilOrder(message.orderId); // these run in parallel, up to 4 at a time
});

// concurrency <= 1 behaves exactly like createQueue. For CPU-bound work prefer Web Workers.

// typed-notifications.ts
import { z } from 'zod';
import { createTypedQueue } from '@netscript/queue';

const NotificationSchema = z.object({
  type: z.enum(['email', 'sms']),
  to: z.string(),
  body: z.string(),
});

// Invalid payloads route to a dead-letter queue instead of being discarded.
const notifications = createTypedQueue('notifications', NotificationSchema, {
  onValidationError: 'dlq',
});

// pg-queue.ts
import { createQueue, QueueProvider } from '@netscript/queue';

// PostgreSQL is never auto-detected — opt in explicitly.
const jobs = createQueue<{ orderId: string }>('order-jobs', {
  provider: QueueProvider.Postgres,        // or provider: 'postgres'
  connection: {
    postgres: {
      // url is optional — falls back to the Aspire-provisioned Postgres URI.
      url: 'postgres://app:secret@localhost:5432/app',
      tableName: 'message_queue',           // default table name
    },
  },
});

await jobs.enqueue({ orderId: 'ord_123' });

// pin-provider.ts
import { createQueue, QueueProvider } from '@netscript/queue';

// Force Redis regardless of what auto-detection would pick.
const jobs = createQueue('jobs', { provider: QueueProvider.Redis });

// Or tune the Deno KV / KV-polling adapter for a remote KV Connect endpoint.
const remote = createQueue('jobs', {
  connection: {
    denoKv: { path: 'https://kv.example.com', pollInterval: 500, visibilityTimeout: 60_000 },
  },
});

// daily-report.ts
import { createScheduler, CronPresets } from '@netscript/cron';

const scheduler = createScheduler(); // native Deno.cron() when available

const report = async () => {
  await generateDailyReport();
};

await scheduler.schedule('daily-report', CronPresets.WEEKDAYS_9AM, report, {
  timezone: 'America/New_York',
});

// Fire once on demand — useful for verifying wiring.
await scheduler.trigger('daily-report');
await scheduler.stop();

// hourly-cleanup.ts
import { createScheduler } from '@netscript/cron';

const scheduler = createScheduler({ provider: 'memory', tickInterval: 100 });

// Hourly cleanup; the contextual handler receives scheduling metadata.
const cleanup = async (context) => {
  console.log(context.jobId, context.attempt, context.scheduledTime.toISOString());
};

await scheduler.schedule('cleanup', '0 * * * *', cleanup);

// observe-jobs.ts
import { createScheduler } from '@netscript/cron';

const scheduler = createScheduler({ provider: 'memory' });

// Observe jobRun / jobError / jobScheduled / jobUnscheduled.
scheduler.on('jobRun', (event) => {
  if (!event.result.success) {
    console.error(`Job ${event.jobId} failed`, event.result.error);
  }
});

// validate-cron.ts
import { isValidCronExpression, parseCronExpression } from '@netscript/cron';

if (isValidCronExpression('0 9 * * 1-5')) {
  const parsed = parseCronExpression('0 9 * * 1-5');
  console.log(parsed?.hour, parsed?.dayOfWeek);
}