Hi

\",\n});\n```\n\n## Val Town Utility Functions\n\nVal Town provides several utility functions to help with common project tasks.\n\n### Importing Utilities\n\nAlways import utilities with version pins to avoid breaking changes:\n\n```ts\nimport {\n parseProject,\n readFile,\n serveFile,\n} from \"https://esm.town/v/std/utils@85-main/index.ts\";\n```\n\n### Available Utilities\n\n#### **serveFile** - Serve project files with proper content types\n\nFor example, in Hono:\n\n```ts\n// serve all files in frontend/ and shared/\napp.get(\"/frontend/*\", (c) => serveFile(c.req.path, import.meta.url));\napp.get(\"/shared/*\", (c) => serveFile(c.req.path, import.meta.url));\n```\n\n#### **readFile** - Read files from within the project:\n\n```ts\n// Read a file from the project\nconst fileContent = await readFile(\"/frontend/index.html\", import.meta.url);\n```\n\n#### **listFiles** - List all files in the project\n\n```ts\nconst files = await listFiles(import.meta.url);\n```\n\n#### **parseProject** - Extract information about the current project from import.meta.url\n\nThis is useful for including info for linking back to a val, ie in \"view source\"\nurls:\n\n```ts\nconst projectVal = parseProject(import.meta.url);\nconsole.log(projectVal.username); // Owner of the project\nconsole.log(projectVal.name); // Project name\nconsole.log(projectVal.version); // Version number\nconsole.log(projectVal.branch); // Branch name\nconsole.log(projectVal.links.self.project); // URL to the project page\n```\n\nHowever, it's _extremely importing_ to note that `parseProject` and other\nStandard Library utilities ONLY RUN ON THE SERVER. If you need access to this\ndata on the client, run it in the server and pass it to the client by splicing\nit into the HTML page or by making an API request for it.\n\n## Val Town Platform Specifics\n\n- **Redirects:** Use\n `return new Response(null, { status: 302, headers: { Location: \"/place/to/redirect\" }})`\n instead of `Response.redirect` which is broken\n- **Images:** Avoid external images or base64 images. Use emojis, unicode\n symbols, or icon fonts/libraries instead\n- **AI Image:** To inline generate an AI image use:\n `

`\n- **Storage:** DO NOT use the Deno KV module for storage\n- **Browser APIs:** DO NOT use the `alert()`, `prompt()`, or `confirm()` methods\n- **Weather Data:** Use open-meteo for weather data (doesn't require API keys)\n unless otherwise specified\n- **View Source:** Add a view source link by importing & using\n `import.meta.url.replace(\"ems.sh\", \"val.town)\"` (or passing this data to the\n client) and include `target=\"_top\"` attribute\n- **Error Debugging:** Add\n `` to HTML to capture\n client-side errors\n- **Error Handling:** Only use try...catch when there's a clear local\n resolution; Avoid catches that merely log or return 500s. Let errors bubble up\n with full context\n- **Environment Variables:** Use `Deno.env.get('keyname')` when you need to, but\n generally prefer APIs that don't require keys\n- **Imports:** Use `https://esm.sh` for npm and Deno dependencies to ensure\n compatibility on server and browser\n- **Storage Strategy:** Only use backend storage if explicitly required; prefer\n simple static client-side sites\n- **React Configuration:** When using React libraries, pin versions with\n `?deps=react@18.2.0,react-dom@18.2.0` and start the file with\n `/** @jsxImportSource https://esm.sh/react@18.2.0 */`\n- Ensure all React dependencies and sub-dependencies are pinned to the same\n version\n- **Styling:** Default to using TailwindCSS via\n `` unless otherwise\n specified\n\n## Project Structure and Design Patterns\n\n### Recommended Directory Structure\n\n```\n├── backend/\n│ ├── database/\n│ │ ├── migrations.ts # Schema definitions\n│ │ ├── queries.ts # DB query functions\n│ │ └── README.md\n│ └── routes/ # Route modules\n│ ├── [route].ts\n│ └── static.ts # Static file serving\n│ ├── index.ts # Main entry point\n│ └── README.md\n├── frontend/\n│ ├── components/\n│ │ ├── App.tsx\n│ │ └── [Component].tsx\n│ ├── favicon.svg\n│ ├── index.html # Main HTML template\n│ ├── index.tsx # Frontend JS entry point\n│ ├── README.md\n│ └── style.css\n├── README.md\n└── shared/\n ├── README.md\n └── utils.ts # Shared types and functions\n```\n\n### Backend (Hono) Best Practices\n\n- Hono is the recommended API framework\n- Main entry point should be `backend/index.ts`\n- **Static asset serving:** Use the utility functions to read and serve project\n files:\n\n ```ts\n import {\n readFile,\n serveFile,\n } from \"https://esm.town/v/std/utils@85-main/index.ts\";\n\n // serve all files in frontend/ and shared/\n app.get(\"/frontend/*\", (c) => serveFile(c.req.path, import.meta.url));\n app.get(\"/shared/*\", (c) => serveFile(c.req.path, import.meta.url));\n\n // For index.html, often you'll want to bootstrap with initial data\n app.get(\"/\", async (c) => {\n let html = await readFile(\"/frontend/index.html\", import.meta.url);\n\n // Inject data to avoid extra round-trips\n const initialData = await fetchInitialData();\n const dataScript = ``;\n\n html = html.replace(\"\", `${dataScript}`);\n return c.html(html);\n });\n ```\n\n- Create RESTful API routes for CRUD operations\n- Always include this snippet at the top-level Hono app to re-throwing errors to\n see full stack traces:\n ```ts\n // Unwrap Hono errors to see original error details\n app.onError((err, c) => {\n throw err;\n });\n ```\n\n### Database Patterns\n\n- Run migrations on startup or comment out for performance\n- Change table names when modifying schemas rather than altering\n- Export clear query functions with proper TypeScript typing\n\n## Common Gotchas and Solutions\n\n1. **Environment Limitations:**\n\n - Val Town runs on Deno in a serverless context, not Node.js\n - Code in `shared/` must work in both frontend and backend environments\n - Cannot use `Deno` keyword in shared code\n - Use `https://esm.sh` for imports that work in both environments\n\n2. **SQLite Peculiarities:**\n\n - Limited support for ALTER TABLE operations\n - Create new tables with updated schemas and copy data when needed\n - Always run table creation before querying\n\n3. **React Configuration:**\n\n - All React dependencies must be pinned to 18.2.0\n - Always include `@jsxImportSource https://esm.sh/react@18.2.0` at the top of\n React files\n - Rendering issues often come from mismatched React versions\n\n4. **File Handling:**\n\n - Val Town only supports text files, not binary\n - Use the provided utilities to read files across branches and forks\n - For files in the project, use `readFile` helpers\n\n5. **API Design:**\n - `fetch` handler is the entry point for HTTP vals\n - Run the Hono app with\n `export default app.fetch // This is the entry point for HTTP vals`\n",[],[174],{"_104":183,"_95":184,"_154":185,"_156":-5,"_157":139,"_158":186,"_160":161,"_106":187,"_100":188,"_163":164,"_165":189,"_167":22,"_168":-5,"_169":190,"_98":191},"182eee7c-c8e8-11f0-9a3b-42dde27851f2","QUICKSTART.md","955f6422-c8ce-11f0-9c36-42dde27851f2",51,["D",1763955961178],3,"# Quick Start Guide\n\nA lightweight SQLite-based queue system for Val Town with priority support, scheduled execution, retries, and monitoring.\n\n## 🎯 What You Get\n\n- ✅ SQLite-backed task queue with ACID guarantees\n- ✅ Priority-based processing\n- ✅ Delayed/scheduled task execution\n- ✅ Automatic retry logic with configurable max attempts\n- ✅ Dead letter queue for failed tasks\n- ✅ Worker processing tasks every minute\n- ✅ Built-in monitoring endpoint\n\n## 🚀 Getting Started\n\n### 1. Initialize the Database\n\n```bash\ncurl https://lqw.web.val.run/schema\n```\n\nThis creates the `queue` and `dead_letter_queue` tables.\n\n### 2. Enqueue Your First Task\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"log_message\",\n \"payload\": {\n \"message\": \"Hello from the queue!\",\n \"level\": \"info\"\n }\n }'\n```\n\nResponse:\n\n```json\n{\n \"success\": true,\n \"task_id\": \"1\",\n \"status\": \"queued\",\n \"scheduled_for\": null\n}\n```\n\n### 3. Monitor the Queue\n\n```bash\ncurl https://lqw.web.val.run/monitor\n```\n\nView stats on task counts, processing durations, dead letter queue, and upcoming scheduled tasks.\n\n## 📋 Available Task Types\n\n| Type | Purpose | Payload |\n| ---------------- | ----------------------- | ---------------------------------------------------------------- |\n| **log_message** | Log to console | `{ message: string, level: \"info\" \\| \"warning\" \\| \"error\" }` |\n| **send_email** | Send email via Val Town | `{ to: string, subject: string, text: string }` |\n| **webhook** | POST to an external URL | `{ url: string, method?: string, body?: any, headers?: object }` |\n| **http_request** | Make HTTP requests | `{ url: string, method?: string, body?: any, headers?: object }` |\n\n## 💡 Common Examples\n\n### High Priority Task\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"log_message\",\n \"payload\": { \"message\": \"Urgent!\", \"level\": \"info\" },\n \"priority\": 10\n }'\n```\n\n### Scheduled Task (1 hour from now)\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"log_message\",\n \"payload\": { \"message\": \"See you later!\", \"level\": \"info\" },\n \"scheduled_for\": '$(date -v+1H +%s)'\n }'\n```\n\n### Send Email\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"send_email\",\n \"payload\": {\n \"to\": \"user@example.com\",\n \"subject\": \"Queue Test\",\n \"text\": \"Hello from the queue!\"\n }\n }'\n```\n\n### Call a Webhook\n\n```bash\n curl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"webhook\",\n \"payload\": {\n \"url\": \"https://example.com/webhook\",\n \"method\": \"POST\",\n \"body\": { \"event\": \"task_completed\" }\n }\n }'\n```\n\n## 📁 File Structure\n\n| File | Purpose |\n| --------------- | -------------------------------------------------- |\n| `api.ts` | HTTP endpoints (`/schema`, `/enqueue`, `/monitor`) |\n| `schema.ts` | Database schema initialization |\n| `types.ts` | TypeScript types and Zod schemas |\n| `processors.ts` | Task execution logic for each task type |\n| `worker.ts` | Cron that processes queued tasks |\n\n## 🔧 Customizing Task Types\n\nAdd a new task type in `processors.ts`:\n\n1. Add to the `ProcessorMap` type\n2. Create a processor function (e.g., `processMyTask`)\n3. Add it to the `processors` object\n4. Add a Zod schema to `types.ts`\n5. Update `VALID_TASK_TYPES` in `api.ts`\n\nExample:\n\n```typescript\n// In processors.ts\nasync function processMyTask(payload: MyTaskPayload) {\n // Your logic here\n}\n\n// Add to ProcessorMap type and processors object\nconst processors = {\n // ... existing\n my_task: processMyTask,\n};\n```\n\n## ⚙️ Configuration\n\n### Enqueue Parameters\n\n- `task_type` (required): One of the valid task types\n- `payload` (required): Task-specific data\n- `priority` (optional, default: 0): Higher = processed first\n- `max_attempts` (optional, default: 3): Retries before DLQ\n- `scheduled_for` (optional): Unix timestamp for delayed execution\n\n### Worker Behavior\n\n- Processes up to 10 tasks per worker run\n- Runs every minute (configure in Val Town)\n- Respects `priority` and `scheduled_for` when selecting tasks\n- Automatically retries failed tasks up to `max_attempts` times\n- Moves tasks to dead letter queue after max retries\n\n## 📊 Monitoring\n\nThe `/monitor` endpoint returns:\n\n- Task counts by status (pending, processing, completed, failed)\n- Average processing duration per status\n- Breakdown by task type\n- Dead letter queue count and recent failures\n- Upcoming scheduled tasks\n\n## 🔍 Next Steps\n\n1. Read the full `README.md` for detailed API reference\n2. Check `types.ts` for all type definitions\n3. Customize `processors.ts` with your own task types\n4. Use the `/monitor` endpoint to track queue health\n",[],[184],{"_104":193,"_95":194,"_154":195,"_156":-5,"_157":139,"_158":196,"_160":161,"_106":197,"_100":198,"_163":164,"_165":199,"_167":22,"_168":-5,"_169":200,"_98":201},"1831e6a4-c8e8-11f0-8f47-42dde27851f2","README.md","41df3cf6-c8cd-11f0-83d5-42dde27851f2",52,["D",1763955961216],5,"# Lightweight Queue System\n\nA SQLite-based task queue for Val Town with priority scheduling, delayed execution, automatic retries, and dead letter queue support. Simple, reliable, and production-ready.\n\n## Features\n\n- ✅ SQLite-backed queue with ACID guarantees\n- ✅ Priority-based task processing\n- ✅ Delayed/scheduled task execution\n- ✅ Automatic retry logic with configurable max attempts\n- ✅ Dead letter queue for failed tasks\n- ✅ Task status tracking (pending, processing, completed, failed)\n- ✅ Built-in monitoring API\n- ✅ Type-safe with TypeScript and Zod validation\n- ✅ Worker processes up to 10 tasks per minute\n\n## Quick Start\n\n### 1. Initialize the Database\n\n```bash\ncurl https://lqw.web.val.run/schema\n```\n\n### 2. Add Your First Task\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"log_message\",\n \"payload\": {\n \"message\": \"Hello queue!\",\n \"level\": \"info\"\n }\n }'\n```\n\n### 3. Check Queue Status\n\n```bash\ncurl https://lqw.web.val.run/monitor\n```\n\n## API Reference\n\n### POST `/enqueue`\n\nEnqueue a new task.\n\n**Request Body:**\n\n```json\n{\n \"task_type\": \"log_message\",\n \"payload\": { \"message\": \"Hello\", \"level\": \"info\" },\n \"priority\": 0,\n \"max_attempts\": 3,\n \"scheduled_for\": null\n}\n```\n\n**Parameters:**\n\n| Parameter | Type | Default | Description |\n| --------------- | ------ | -------- | -------------------------------------------------------------- |\n| `task_type` | string | required | One of: `log_message`, `send_email`, `webhook`, `http_request` |\n| `payload` | object | required | Task-specific data (structure depends on task_type) |\n| `priority` | number | 0 | Priority level (higher = processed first) |\n| `max_attempts` | number | 3 | Max retry attempts before moving to DLQ |\n| `scheduled_for` | number | null | Unix timestamp for delayed execution |\n\n**Response:**\n\n```json\n{\n \"success\": true,\n \"task_id\": \"1\",\n \"status\": \"queued\",\n \"scheduled_for\": null\n}\n```\n\n### GET `/schema`\n\nInitialize or check the database schema.\n\n**Response:**\n\n```json\n{\n \"success\": true,\n \"message\": \"Queue schema initialized successfully!\",\n \"tables\": [\"queue\", \"dead_letter_queue\"]\n}\n```\n\n### GET `/monitor`\n\nGet queue statistics and monitoring data.\n\n**Response:**\n\n```json\n{\n \"timestamp\": \"2024-11-23T10:00:00.000Z\",\n \"stats\": {\n \"pending\": { \"count\": 5, \"avg_duration_seconds\": null },\n \"processing\": { \"count\": 1, \"avg_duration_seconds\": 2.3 },\n \"completed\": { \"count\": 42, \"avg_duration_seconds\": 1.5 },\n \"failed\": { \"count\": 0, \"avg_duration_seconds\": null }\n },\n \"task_types\": {\n \"log_message\": { \"count\": 30 },\n \"send_email\": { \"count\": 15 },\n \"webhook\": { \"count\": 2 },\n \"http_request\": { \"count\": 1 }\n },\n \"dead_letter_queue\": {\n \"count\": 2,\n \"recent\": [\n {\n \"id\": 1,\n \"task_type\": \"send_email\",\n \"error\": \"Invalid email address\",\n \"attempts\": 3,\n \"created_at\": \"2024-11-23T09:00:00.000Z\"\n }\n ]\n },\n \"scheduled_tasks\": [\n {\n \"id\": 5,\n \"task_type\": \"log_message\",\n \"scheduled_for\": \"2024-11-23T11:00:00.000Z\",\n \"priority\": 0\n }\n ]\n}\n```\n\n## Task Types\n\n### log_message\n\nLog a message to the console.\n\n```json\n{\n \"task_type\": \"log_message\",\n \"payload\": {\n \"message\": \"Something happened\",\n \"level\": \"info\"\n }\n}\n```\n\n**Payload:**\n\n| Field | Type | Options | Description |\n| --------- | ------ | -------------------------------- | -------------- |\n| `message` | string | required | Message to log |\n| `level` | string | `\"info\"`, `\"warning\"`, `\"error\"` | Log level |\n\n### send_email\n\nSend an email via Val Town's email service.\n\n```json\n{\n \"task_type\": \"send_email\",\n \"payload\": {\n \"to\": \"user@example.com\",\n \"subject\": \"Hello\",\n \"text\": \"Email body\"\n }\n}\n```\n\n**Payload:**\n\n| Field | Type | Description |\n| --------- | ------ | ----------------------- |\n| `to` | string | Recipient email address |\n| `subject` | string | Email subject |\n| `text` | string | Email body text |\n\n### webhook\n\nCall an external webhook endpoint.\n\n```json\n{\n \"task_type\": \"webhook\",\n \"payload\": {\n \"url\": \"https://example.com/webhook\",\n \"method\": \"POST\",\n \"body\": { \"event\": \"task_complete\" },\n \"headers\": { \"Authorization\": \"Bearer token123\" }\n }\n}\n```\n\n**Payload:**\n\n| Field | Type | Default | Description |\n| --------- | ------ | -------- | ------------------- |\n| `url` | string | required | Webhook URL |\n| `method` | string | `\"POST\"` | HTTP method |\n| `body` | object | null | Request body (JSON) |\n| `headers` | object | `{}` | HTTP headers |\n\n### http_request\n\nMake an HTTP request to any endpoint.\n\n```json\n{\n \"task_type\": \"http_request\",\n \"payload\": {\n \"url\": \"https://api.example.com/data\",\n \"method\": \"GET\",\n \"headers\": { \"Authorization\": \"Bearer token\" }\n }\n}\n```\n\n**Payload:**\n\n| Field | Type | Default | Description |\n| --------- | ------ | -------- | ------------ |\n| `url` | string | required | Request URL |\n| `method` | string | `\"GET\"` | HTTP method |\n| `body` | object | null | Request body |\n| `headers` | object | `{}` | HTTP headers |\n\n## Examples\n\n### High Priority Task\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"log_message\",\n \"payload\": { \"message\": \"Urgent!\", \"level\": \"error\" },\n \"priority\": 100\n }'\n```\n\nTasks with higher priority values are processed first.\n\n### Schedule for Later\n\n```bash\n# Schedule for 1 hour from now\nSCHEDULED_TIME=$(($(date +%s) + 3600))\n\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d \"{\n \\\"task_type\\\": \\\"log_message\\\",\n \\\"payload\\\": { \\\"message\\\": \\\"Scheduled message\\\", \\\"level\\\": \\\"info\\\" },\n \\\"scheduled_for\\\": $SCHEDULED_TIME\n }\"\n```\n\n### Configure Retries\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"webhook\",\n \"payload\": {\n \"url\": \"https://example.com/webhook\",\n \"method\": \"POST\",\n \"body\": { \"data\": \"value\" }\n },\n \"max_attempts\": 5\n }'\n```\n\nThis task will be retried up to 5 times before moving to the dead letter queue.\n\n### Send Email with Queue\n\n```bash\ncurl -X POST https://lqw.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"send_email\",\n \"payload\": {\n \"to\": \"user@example.com\",\n \"subject\": \"Welcome!\",\n \"text\": \"Thanks for signing up\"\n }\n }'\n```\n\n## Adding Custom Task Types\n\nAdding a new task type requires changes to three files: `types.ts`, `processors.ts`, and `api.ts`. Follow this step-by-step guide:\n\n### Step 1: Define the Payload Type in `types.ts`\n\nAdd a Zod schema to validate your task payload. This ensures type safety and runtime validation:\n\n```typescript\nexport const slackNotificationPayloadSchema = z.object({\n webhook_url: z.string().url(),\n channel: z.string(),\n message: z.string(),\n emoji: z.string().optional().default(\":robot_face:\"),\n});\n\nexport type SlackNotificationPayload = z.infer<\n typeof slackNotificationPayloadSchema\n>;\n```\n\n**Key points:**\n\n- Use `.url()`, `.email()`, etc. for common patterns\n- Use `.optional()` for optional fields and `.default()` for defaults\n- Export both the schema and the inferred type\n\n### Step 2: Create the Processor in `processors.ts`\n\nCreate an async function that processes the task. This function is called by the worker:\n\n```typescript\nasync function processSlackNotification(payload: SlackNotificationPayload) {\n const { webhook_url, channel, message, emoji } = payload;\n\n const response = await fetch(webhook_url, {\n method: \"POST\",\n headers: { \"Content-Type\": \"application/json\" },\n body: JSON.stringify({\n channel,\n text: message,\n icon_emoji: emoji,\n }),\n });\n\n if (!response.ok) {\n throw new Error(\n `Slack webhook failed: ${response.status} ${response.statusText}`\n );\n }\n\n console.log(`Slack message sent to ${channel}`);\n}\n```\n\n**Best practices:**\n\n- Always validate external responses with `if (!response.ok)`\n- Throw descriptive errors - they'll be logged and trigger retries\n- Log success for monitoring\n- Handle timeouts with explicit fetch timeouts: `{ timeout: 10000 }`\n\n### Step 3: Register in the ProcessorMap\n\nUpdate the `ProcessorMap` type and `processors` object in `processors.ts`:\n\n```typescript\ntype ProcessorMap = {\n send_email: EmailPayload;\n webhook: WebhookPayload;\n log_message: LogMessagePayload;\n http_request: HttpRequestPayload;\n slack_notification: SlackNotificationPayload; // Add here\n};\n\nconst payloadSchemas = {\n send_email: emailPayloadSchema,\n webhook: webhookPayloadSchema,\n log_message: logMessagePayloadSchema,\n http_request: httpRequestPayloadSchema,\n slack_notification: slackNotificationPayloadSchema, // Add here\n} as const;\n\nconst processors: {\n [K in TaskType]: Processor;\n} = {\n send_email: processSendEmail,\n webhook: processWebhook,\n log_message: processLogMessage,\n http_request: processHttpRequest,\n slack_notification: processSlackNotification, // Add here\n};\n```\n\n### Step 4: Update VALID_TASK_TYPES in `api.ts`\n\nAdd your task type to the `VALID_TASK_TYPES` array:\n\n```typescript\nexport const VALID_TASK_TYPES = [\n \"send_email\",\n \"webhook\",\n \"log_message\",\n \"http_request\",\n \"slack_notification\", // Add here\n] as const;\n```\n\n### Testing Your Task Type\n\nOnce added, test by enqueueing a task:\n\n```bash\ncurl -X POST https://your-val.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"slack_notification\",\n \"payload\": {\n \"webhook_url\": \"https://hooks.slack.com/services/YOUR/WEBHOOK/URL\",\n \"channel\": \"#notifications\",\n \"message\": \"Test message\",\n \"emoji\": \":bell:\"\n }\n }'\n```\n\nThen check the monitor to see if it succeeded:\n\n```bash\ncurl https://your-val.web.val.run/monitor | jq '.task_types'\n```\n\n### Common Patterns\n\n**Task with retryable external API:**\n\n```typescript\nasync function processDataSync(payload: DataSyncPayload) {\n const response = await fetch(payload.api_url, {\n timeout: 30000,\n headers: payload.headers,\n });\n\n // 5xx errors should retry; 4xx should fail permanently\n if (response.status >= 500) {\n throw new Error(`API error: ${response.status}`);\n }\n\n if (!response.ok) {\n throw new Error(`Data sync failed: ${response.status} - won't retry`);\n }\n\n const data = await response.json();\n console.log(\"Data synced successfully\");\n}\n```\n\n**Task that updates local state:**\n\n```typescript\nasync function processDataAggregation(payload: DataAggregationPayload) {\n // Fetch data\n const results = await Promise.all(\n payload.sources.map((url) => fetch(url).then((r) => r.json()))\n );\n\n // Store aggregated result (e.g., in blob storage)\n // await blob.setJSON(`aggregation_${Date.now()}`, results);\n\n console.log(\"Aggregation complete\");\n}\n```\n\n**Task with retry-limiting logic:**\n\n```typescript\nasync function processExpensiveOperation(payload: ExpensiveOpPayload) {\n // Some tasks you only want to retry a few times\n if (payload.max_retries === 0) {\n // Handle no-retry case\n console.log(\"One-shot operation\");\n }\n\n // Your expensive logic here\n}\n```\n\n### Validation and Error Messages\n\nThe Zod schemas automatically validate payloads. Invalid payloads get rejected at enqueue time:\n\n```bash\n# This will fail with validation error (missing required fields)\ncurl -X POST https://your-val.web.val.run/enqueue \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"task_type\": \"slack_notification\",\n \"payload\": {\n \"channel\": \"#notifications\"\n # Missing webhook_url and message\n }\n }'\n```\n\nResponse:\n\n```json\n{\n \"error\": \"Validation error: webhook_url is required\"\n}\n```\n\n### Debugging Task Types\n\nIf your task fails:\n\n1. **Check dead letter queue** via `/monitor` endpoint\n2. **Look for error messages** in the recent failures section\n3. **Check Val Town logs** for console output\n4. **Test manually** by calling your processor function directly\n5. **Use type-safety** - TypeScript will catch most issues at dev time\n\n### Checklist for New Task Types\n\n- [ ] Zod schema defined in `types.ts` with all required fields\n- [ ] Type exported from `types.ts`\n- [ ] Processor function added to `processors.ts`\n- [ ] Error handling: throws descriptive errors on failure\n- [ ] Type added to `ProcessorMap`\n- [ ] Schema added to `payloadSchemas`\n- [ ] Processor added to `processors` object\n- [ ] Task type added to `VALID_TASK_TYPES` in `api.ts`\n- [ ] Tested with `/enqueue` endpoint\n- [ ] Verified in `/monitor` endpoint stats\n\n## Architecture\n\n| File | Purpose |\n| --------------- | -------------------------------------------------------------- |\n| `api.ts` | HTTP handler with endpoints: `/schema`, `/enqueue`, `/monitor` |\n| `schema.ts` | Database schema initialization and validation |\n| `types.ts` | TypeScript types and Zod validation schemas |\n| `processors.ts` | Task execution logic for each task type |\n| `worker.ts` | Worker that processes queued tasks (runs every minute) |\n\n## Database Schema\n\n### `queue` Table\n\n| Column | Type | Description |\n| --------------- | ------- | --------------------------------------------------- |\n| `id` | INTEGER | Primary key |\n| `task_type` | TEXT | Task type (e.g., \"send_email\") |\n| `payload` | TEXT | JSON-encoded task data |\n| `status` | TEXT | Task status: pending, processing, completed, failed |\n| `created_at` | INTEGER | Unix timestamp |\n| `started_at` | INTEGER | Unix timestamp when processing started |\n| `completed_at` | INTEGER | Unix timestamp when task completed |\n| `scheduled_for` | INTEGER | Unix timestamp for delayed execution |\n| `attempts` | INTEGER | Number of attempts made |\n| `max_attempts` | INTEGER | Maximum attempts before DLQ |\n| `priority` | INTEGER | Task priority (higher = first) |\n| `error` | TEXT | Error message from last failed attempt |\n\n### `dead_letter_queue` Table\n\nContains tasks that failed after exhausting all retry attempts. Has the same columns as `queue` plus:\n\n| Column | Type | Description |\n| ----------------- | ------- | -------------------------------- |\n| `moved_to_dlq_at` | INTEGER | Unix timestamp when moved to DLQ |\n\n## Worker Behavior\n\nThe worker runs every minute and:\n\n1. Selects up to 10 pending tasks sorted by priority and creation time\n2. Skips tasks with `scheduled_for` in the future\n3. Marks task as \"processing\"\n4. Executes the task processor\n5. On success: marks task as \"completed\"\n6. On failure:\n - If attempts < max_attempts: resets status to \"pending\" for retry\n - If attempts >= max_attempts: moves task to dead letter queue\n\n## Error Handling\n\n- Task failures are logged with error messages\n- Failed tasks are automatically retried based on `max_attempts`\n- After max retries, tasks are moved to the dead letter queue\n- The `/monitor` endpoint shows recent failures for debugging\n\n## Best Practices\n\n1. **Use meaningful task types** - makes monitoring easier\n2. **Set reasonable priorities** - avoid inversions\n3. **Configure max_attempts** - based on your tolerance for retries\n4. **Monitor the dead letter queue** - investigate failures\n5. **Use scheduled_for** - for distributed task timing\n6. **Keep payloads small** - SQLite isn't a document DB\n7. **Validate payloads** - Zod schemas catch errors early\n\n## Limits\n\n- Maximum 10 tasks processed per minute\n- SQLite row size limit ~1GB (payloads should be much smaller)\n- Task payloads are JSON-serialized strings\n",[],[194],{"_104":203,"_95":99,"_154":204,"_156":-5,"_157":139,"_158":101,"_160":22,"_106":205,"_100":206,"_163":207,"_165":208,"_167":22,"_168":209,"_169":210,"_98":211},"344eaa92-c8e9-11f0-af87-42dde27851f2","cfeb2b1c-c8d8-11f0-8358-42dde27851f2",["D",1763956437888],20,"http","import { Hono } from \"https://esm.sh/hono@latest\";\nimport { sqlite } from \"https://esm.town/v/std/sqlite\";\nimport { checkSchema, createSchema } from \"./schema.ts\";\nimport { type TaskType, VALID_TASK_TYPES } from \"./types.ts\";\n\nconst app = new Hono();\n\napp.get(\"/\", (c) => {\n return c.json({\n \"GET /schema\": \"initialize or check database schema\",\n \"POST /enqueue\": \"enqueue a task\",\n \"GET /monitor\": \"monitor your tasks\",\n });\n});\n\napp.get(\"/schema\", async (c) => {\n try {\n const exists = await checkSchema();\n if (exists) {\n return c.json({ success: true, message: \"Queue schema already exists\" });\n }\n const result = await createSchema();\n return c.json(result);\n } catch (error) {\n if (error instanceof Error) {\n return c.json({ success: false, error: error.message }, 500);\n }\n return c.json(\n {\n success: false,\n error: `Unknown error occurred during schema initialization: ${error}`,\n },\n 500,\n );\n }\n});\n\napp.post(\"/enqueue\", async (c) => {\n const body = await c.req.json();\n const {\n task_type,\n payload,\n priority = 0,\n max_attempts = 3,\n scheduled_for = null,\n } = body;\n\n try {\n if (!task_type) {\n return c.json({ error: \"task_type is required\" }, 400);\n }\n\n if (!VALID_TASK_TYPES.includes(task_type)) {\n return Response.json(\n {\n error: `Invalid task_type. Must be one of: ${\n VALID_TASK_TYPES.join(\", \")\n }`,\n valid_types: VALID_TASK_TYPES,\n },\n { status: 400 },\n );\n }\n\n if (payload === undefined) {\n return c.json({ error: \"payload is required\" }, 400);\n }\n\n const result = await sqlite.execute({\n sql:\n `INSERT INTO queue (task_type, payload, priority, max_attempts, scheduled_for)\n VALUES (?, ?, ?, ?, ?)`,\n args: [\n task_type,\n JSON.stringify(payload),\n priority,\n max_attempts,\n scheduled_for,\n ],\n });\n\n return c.json({\n success: true,\n task_id: result.lastInsertRowid?.toString(),\n status: scheduled_for ? \"scheduled\" : \"queued\",\n scheduled_for: scheduled_for\n ? new Date(scheduled_for * 1000).toISOString()\n : null,\n });\n } catch (error) {\n console.error(\"Error enqueueing task:\", error);\n return c.json({ error: error.message }, 500);\n }\n});\n\napp.get(\"/monitor\", async (c) => {\n try {\n // Get overall stats\n const statsResult = await sqlite.execute(`\n SELECT\n status,\n COUNT(*) as count,\n AVG(CASE\n WHEN completed_at IS NOT NULL AND started_at IS NOT NULL\n THEN completed_at - started_at\n ELSE NULL\n END) as avg_duration_seconds\n FROM queue\n GROUP BY status\n `);\n\n const stats: Record<\n string,\n { count: number; avg_duration_seconds: number | null }\n > = {};\n for (const row of statsResult.rows) {\n stats[String(row[0])] = {\n count: Number(row[1]),\n avg_duration_seconds: row[2]\n ? Math.round(Number(row[2]) * 100) / 100\n : null,\n };\n }\n\n // Get task type breakdown\n const taskTypesResult = await sqlite.execute(`\n SELECT task_type, COUNT(*) as count\n FROM queue\n GROUP BY task_type\n ORDER BY count DESC\n `);\n\n const taskTypes: Record = taskTypesResult.rows\n .reduce(\n (acc, row) => {\n acc[String(row[0]) as TaskType] = { count: Number(row[1]) };\n return acc;\n },\n {} as Record,\n );\n\n const dlqResult = await sqlite.execute(\n `SELECT COUNT(*) as count FROM dead_letter_queue`,\n );\n\n const dlqCount = Number(dlqResult.rows[0][0]);\n\n const recentFailedResult = await sqlite.execute(`\n SELECT id, task_type, error, attempts, created_at\n FROM dead_letter_queue\n ORDER BY moved_to_dlq_at DESC\n LIMIT 5\n `);\n\n const recentFailed: {\n id: number;\n task_type: TaskType;\n error: string;\n attempts: number;\n created_at: string | null;\n }[] = recentFailedResult.rows.map((row) => ({\n id: Number(row[0]),\n task_type: String(row[1]) as TaskType,\n error: String(row[2] || \"\"),\n attempts: Number(row[3]),\n created_at: row[4] ? new Date(Number(row[4]) * 1000).toISOString() : null,\n }));\n\n // Get upcoming scheduled tasks\n const scheduledResult = await sqlite.execute(`\n SELECT id, task_type, scheduled_for, priority\n FROM queue\n WHERE scheduled_for IS NOT NULL AND status = 'pending'\n ORDER BY scheduled_for ASC\n LIMIT 10\n `);\n\n const scheduled: {\n id: number;\n task_type: TaskType;\n scheduled_for: string | null;\n priority: number;\n }[] = scheduledResult.rows.map((row) => ({\n id: Number(row[0]),\n task_type: String(row[1]) as TaskType,\n scheduled_for: row[2]\n ? new Date(Number(row[2]) * 1000).toISOString()\n : null,\n priority: Number(row[3]),\n }));\n\n return c.json(\n {\n timestamp: new Date().toISOString(),\n stats,\n task_types: taskTypes,\n dead_letter_queue: {\n count: dlqCount,\n recent: recentFailed,\n },\n scheduled_tasks: scheduled,\n },\n 200,\n );\n } catch (error) {\n console.error(\"Error fetching monitor data:\", error);\n return c.json({ error: error.message }, 500);\n }\n});\n\nexport default app.fetch;","lqw",[],[99],{"_104":213,"_95":214,"_154":215,"_156":-5,"_157":139,"_158":216,"_160":161,"_106":217,"_100":32,"_163":164,"_165":218,"_167":22,"_168":-5,"_169":219,"_98":220},"f6ddc4b6-c8d1-11f0-91f3-42dde27851f2","biome.json","f6ddc588-c8d1-11f0-91f3-42dde27851f2",26,["D",1763946456231],"{\n\t\"$schema\": \"https://biomejs.dev/schemas/2.3.7/schema.json\",\n\t\"vcs\": {\n\t\t\"enabled\": false,\n\t\t\"clientKind\": \"git\",\n\t\t\"useIgnoreFile\": false\n\t},\n\t\"files\": {\n\t\t\"ignoreUnknown\": false\n\t},\n\t\"formatter\": {\n\t\t\"enabled\": true,\n\t\t\"indentStyle\": \"tab\"\n\t},\n\t\"linter\": {\n\t\t\"enabled\": true,\n\t\t\"rules\": {\n\t\t\t\"recommended\": true\n\t\t}\n\t},\n\t\"javascript\": {\n\t\t\"formatter\": {\n\t\t\t\"quoteStyle\": \"double\"\n\t\t}\n\t},\n\t\"assist\": {\n\t\t\"enabled\": true,\n\t\t\"actions\": {\n\t\t\t\"source\": {\n\t\t\t\t\"organizeImports\": \"on\"\n\t\t\t}\n\t\t}\n\t}\n}\n",[],[214],{"_104":222,"_95":223,"_154":224,"_156":-5,"_157":139,"_158":225,"_160":161,"_106":226,"_100":227,"_163":164,"_165":228,"_167":22,"_168":-5,"_169":229,"_98":230},"52b5acd4-c8e8-11f0-8fbe-42dde27851f2","deno.json","f6c357b6-c8d1-11f0-99b0-42dde27851f2",55,["D",1763956059377],4,"{\n\t\"$schema\": \"https://raw.githubusercontent.com/denoland/deno/348900b8b79f4a434cab4c74b3bc8d4d2fa8ee74/cli/schemas/config-file.v1.json\",\n\t\"compilerOptions\": {\n\t\t\"lib\": [\n\t\t\t\"dom\",\n\t\t\t\"dom.iterable\",\n\t\t\t\"dom.asynciterable\",\n\t\t\t\"deno.ns\",\n\t\t\t\"deno.unstable\"\n\t\t],\n\t\t\"noImplicitAny\": false,\n\t\t\"strict\": false,\n\t\t\"types\": [\"https://www.val.town/types/valtown.d.ts\"]\n\t},\n\t\"experimental\": {\n\t\t\"unstable-node-globals\": true,\n\t\t\"unstable-sloppy-imports\": true,\n\t\t\"unstable-temporal\": true,\n\t\t\"unstable-worker-options\": true\n\t},\n\t\"imports\": {\n\t\t\"@biomejs/biome\": \"npm:@biomejs/biome@^2.3.7\"\n\t},\n\t\"lint\": {\n\t\t\"rules\": {\n\t\t\t\"exclude\": [\"no-explicit-any\", \"no-import-prefix\"]\n\t\t}\n\t},\n\t\"lock\": false,\n\t\"nodeModulesDir\": \"auto\"\n}\n",[],[223],{"_104":232,"_95":233,"_154":234,"_156":-5,"_157":139,"_158":235,"_160":161,"_106":236,"_100":198,"_163":237,"_165":238,"_167":22,"_168":-5,"_169":239,"_98":240},"549c4846-c8e8-11f0-b68e-42dde27851f2","processors.ts","f6681cde-c8cc-11f0-979e-42dde27851f2",56,["D",1763956062586],"script","import { email } from \"https://esm.town/v/std/email\";\nimport type {\n\tEmailPayload,\n\tHttpRequestPayload,\n\tLogMessagePayload,\n\tTask,\n\tTaskType,\n\tWebhookPayload,\n} from \"./types.ts\";\nimport {\n\temailPayloadSchema,\n\thttpRequestPayloadSchema,\n\tlogMessagePayloadSchema,\n\twebhookPayloadSchema,\n} from \"./types.ts\";\n\ntype ProcessorMap = {\n\tsend_email: EmailPayload;\n\twebhook: WebhookPayload;\n\tlog_message: LogMessagePayload;\n\thttp_request: HttpRequestPayload;\n};\n\nconst payloadSchemas = {\n\tsend_email: emailPayloadSchema,\n\twebhook: webhookPayloadSchema,\n\tlog_message: logMessagePayloadSchema,\n\thttp_request: httpRequestPayloadSchema,\n} as const;\n\ntype Processor = (\n\tpayload: ProcessorMap[T],\n) => Promise | void;\n\nconst processors: {\n\t[K in TaskType]: Processor;\n} = {\n\tsend_email: processSendEmail,\n\twebhook: processWebhook,\n\tlog_message: processLogMessage,\n\thttp_request: processHttpRequest,\n};\n\nasync function processTaskByType(\n\ttaskType: T,\n\tpayload: ProcessorMap[T],\n): Promise {\n\tconst processor = processors[taskType];\n\tconst schema = payloadSchemas[taskType];\n\tif (!processor || !schema) {\n\t\tthrow new Error(`Unknown task type: ${taskType}`);\n\t}\n\tconst validatedPayload = schema.parse(payload) as ProcessorMap[T];\n\tawait processor(validatedPayload);\n}\n\nexport async function processTask(task: Task): Promise {\n\tconst { task_type, payload } = task;\n\tconsole.log(`Processing task ${task.id} of type: ${task_type}`);\n\tawait processTaskByType(task_type, payload);\n}\n\nasync function processSendEmail(payload: EmailPayload) {\n\tconst { to, subject, text } = payload;\n\tawait email({ to, subject, text });\n\tconsole.log(`Email sent to ${to}`);\n}\n\nasync function processWebhook(payload: WebhookPayload) {\n\tconst { url, method = \"POST\", body, headers = {} } = payload;\n\n\tconst response = await fetch(url, {\n\t\tmethod,\n\t\theaders: {\n\t\t\t\"Content-Type\": \"application/json\",\n\t\t\t...headers,\n\t\t},\n\t\tbody: body ? JSON.stringify(body) : undefined,\n\t});\n\n\tif (!response.ok) {\n\t\tthrow new Error(\n\t\t\t`Webhook failed: ${response.status} ${response.statusText}`,\n\t\t);\n\t}\n\n\tconsole.log(`Webhook sent to ${url}`);\n}\n\nfunction processLogMessage(payload: LogMessagePayload) {\n\tconst { message, level = \"info\" } = payload;\n\tconsole.log(`[${level.toUpperCase()}] ${message}`);\n}\n\nasync function processHttpRequest(payload: HttpRequestPayload) {\n\tconst { url, method = \"GET\", body, headers = {} } = payload;\n\n\tconst response = await fetch(url, {\n\t\tmethod,\n\t\theaders,\n\t\tbody: body ? JSON.stringify(body) : undefined,\n\t});\n\n\tif (!response.ok) {\n\t\tthrow new Error(\n\t\t\t`HTTP request failed: ${response.status} ${response.statusText}`,\n\t\t);\n\t}\n\n\tconsole.log(`HTTP ${method} request to ${url} completed`);\n}\n",[],[233],{"_104":242,"_95":243,"_154":244,"_156":-5,"_157":139,"_158":245,"_160":161,"_106":246,"_100":198,"_163":207,"_165":247,"_167":22,"_168":-5,"_169":248,"_98":249},"cff09214-c8d8-11f0-a0b0-42dde27851f2","schema.ts","ba92dd0c-c8cc-11f0-9798-42dde27851f2",43,["D",1763949397512],"import { sqlite } from \"https://esm.town/v/std/sqlite\";\n\nexport async function createSchema() {\n\ttry {\n\t\t// Create the main queue table\n\t\tawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS queue (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n task_type TEXT NOT NULL,\n payload TEXT NOT NULL,\n status TEXT DEFAULT 'pending',\n created_at INTEGER DEFAULT (unixepoch()),\n started_at INTEGER,\n completed_at INTEGER,\n scheduled_for INTEGER,\n attempts INTEGER DEFAULT 0,\n max_attempts INTEGER DEFAULT 3,\n priority INTEGER DEFAULT 0,\n error TEXT\n )\n `);\n\n\t\t// Create indexes for efficient queries\n\t\tawait sqlite.execute(`\n CREATE INDEX IF NOT EXISTS idx_status_priority \n ON queue(status, priority DESC, created_at)\n `);\n\n\t\tawait sqlite.execute(`\n CREATE INDEX IF NOT EXISTS idx_scheduled \n ON queue(scheduled_for) \n WHERE scheduled_for IS NOT NULL\n `);\n\n\t\t// Create dead letter queue table\n\t\tawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS dead_letter_queue (\n id INTEGER PRIMARY KEY,\n task_type TEXT NOT NULL,\n payload TEXT NOT NULL,\n status TEXT,\n created_at INTEGER,\n started_at INTEGER,\n completed_at INTEGER,\n attempts INTEGER,\n max_attempts INTEGER,\n priority INTEGER,\n error TEXT,\n moved_to_dlq_at INTEGER DEFAULT (unixepoch())\n )\n `);\n\n\t\tconsole.log(\"Queue schema initialized successfully!\");\n\t\treturn {\n\t\t\tsuccess: true,\n\t\t\tmessage: \"Queue schema initialized successfully!\",\n\t\t\ttables: [\"queue\", \"dead_letter_queue\"],\n\t\t};\n\t} catch (error) {\n\t\tconsole.error(\"Schema initialization error:\", error);\n\t\tthrow error;\n\t}\n}\n\nexport async function checkSchema(): Promise {\n\ttry {\n\t\tconst result = await sqlite.execute(`\n\t\t\tSELECT name FROM sqlite_master WHERE type='table' AND name IN ('queue', 'dead_letter_queue')\n\t\t`);\n\t\treturn result.rows.length > 0;\n\t} catch (error) {\n\t\tconsole.error(\"Schema retrieval error:\", error);\n\t\tthrow error;\n\t}\n}\n",[],[243],{"_104":251,"_95":252,"_154":253,"_156":-5,"_157":139,"_158":254,"_160":161,"_106":255,"_100":256,"_163":237,"_165":257,"_167":22,"_168":-5,"_169":258,"_98":259},"f6d50934-c8d1-11f0-aefb-42dde27851f2","test.ts","3871c642-c8ce-11f0-a873-42dde27851f2",25,["D",1763946456294],2,"/**\n * Test the queue system by enqueueing a few tasks\n */\nexport default async function () {\n\tconst enqueueUrl =\n\t\t\"https://nbbaier--c5bd3e20c8cc11f0a88c42dde27851f2.web.val.run\";\n\n\t// Test 1: Simple log message\n\tconst test1 = await fetch(enqueueUrl, {\n\t\tmethod: \"POST\",\n\t\theaders: { \"Content-Type\": \"application/json\" },\n\t\tbody: JSON.stringify({\n\t\t\ttask_type: \"log_message\",\n\t\t\tpayload: {\n\t\t\t\tmessage: \"Testing the queue system!\",\n\t\t\t\tlevel: \"info\",\n\t\t\t},\n\t\t}),\n\t});\n\n\tconst result1 = await test1.json();\n\tconsole.log(\"Test 1 (log_message):\", result1);\n\n\t// Test 2: High priority task\n\tconst test2 = await fetch(enqueueUrl, {\n\t\tmethod: \"POST\",\n\t\theaders: { \"Content-Type\": \"application/json\" },\n\t\tbody: JSON.stringify({\n\t\t\ttask_type: \"log_message\",\n\t\t\tpayload: {\n\t\t\t\tmessage: \"High priority task!\",\n\t\t\t\tlevel: \"warn\",\n\t\t\t},\n\t\t\tpriority: 10,\n\t\t}),\n\t});\n\n\tconst result2 = await test2.json();\n\tconsole.log(\"Test 2 (high priority):\", result2);\n\n\t// Test 3: Scheduled task (30 seconds from now)\n\tconst scheduledTime = Math.floor(Date.now() / 1000) + 30;\n\tconst test3 = await fetch(enqueueUrl, {\n\t\tmethod: \"POST\",\n\t\theaders: { \"Content-Type\": \"application/json\" },\n\t\tbody: JSON.stringify({\n\t\t\ttask_type: \"log_message\",\n\t\t\tpayload: {\n\t\t\t\tmessage: \"This was scheduled for 30 seconds in the future!\",\n\t\t\t\tlevel: \"info\",\n\t\t\t},\n\t\t\tscheduled_for: scheduledTime,\n\t\t}),\n\t});\n\n\tconst result3 = await test3.json();\n\tconsole.log(\"Test 3 (scheduled):\", result3);\n\n\treturn {\n\t\ttest1: result1,\n\t\ttest2: result2,\n\t\ttest3: result3,\n\t\tmessage:\n\t\t\t\"All tests completed! Check the monitor endpoint to see queue status.\",\n\t};\n}\n",[],[252],{"_104":261,"_95":262,"_154":263,"_156":-5,"_157":139,"_158":264,"_160":161,"_106":265,"_100":178,"_163":237,"_165":266,"_167":22,"_168":-5,"_169":267,"_98":268},"bd5a1cb4-c8e8-11f0-961e-42dde27851f2","types.ts","a6750a1a-c8e6-11f0-943c-42dde27851f2",60,["D",1763956238285],"import { z } from \"https://esm.sh/zod@latest\";\n\nexport const VALID_TASK_TYPES = [\n\t\"send_email\",\n\t\"webhook\",\n\t\"log_message\",\n\t\"http_request\",\n] as const;\n\nexport const taskTypeSchema = z.enum(VALID_TASK_TYPES);\n\nexport const emailPayloadSchema = z.object({\n\tto: z.string(),\n\tsubject: z.string(),\n\ttext: z.string(),\n});\n\nexport const webhookPayloadSchema = z.object({\n\turl: z.string(),\n\tmethod: z.enum([\"GET\", \"POST\", \"PUT\", \"DELETE\"]),\n\tbody: z.any().optional(),\n\theaders: z.record(z.string(), z.string()).optional(),\n});\n\nexport const logMessagePayloadSchema = z.object({\n\tmessage: z.string(),\n\tlevel: z.enum([\"info\", \"warning\", \"error\"]),\n});\n\nexport const httpRequestPayloadSchema = z.object({\n\turl: z.string(),\n\tmethod: z.enum([\"GET\", \"POST\", \"PUT\", \"DELETE\"]),\n\tbody: z.any().optional(),\n\theaders: z.record(z.string(), z.string()).optional(),\n});\n\nexport const payloadSchema = z.union([\n\temailPayloadSchema,\n\twebhookPayloadSchema,\n\tlogMessagePayloadSchema,\n\thttpRequestPayloadSchema,\n]);\n\nexport const taskSchema = z.object({\n\tid: z.number(),\n\ttask_type: taskTypeSchema,\n\tpayload: payloadSchema,\n\tpriority: z.number().optional(),\n\tmax_attempts: z.number().optional(),\n\tscheduled_for: z.number().optional(),\n});\n\nconst emailTaskSchema = taskSchema.extend({\n\ttask_type: z.literal(\"send_email\"),\n\tpayload: emailPayloadSchema,\n});\n\nconst webhookTaskSchema = taskSchema.extend({\n\ttask_type: z.literal(\"webhook\"),\n\tpayload: webhookPayloadSchema,\n});\n\nconst logMessageTaskSchema = taskSchema.extend({\n\ttask_type: z.literal(\"log_message\"),\n\tpayload: logMessagePayloadSchema,\n});\n\nconst httpRequestTaskSchema = taskSchema.extend({\n\ttask_type: z.literal(\"http_request\"),\n\tpayload: httpRequestPayloadSchema,\n});\n\nexport type Task = z.infer;\nexport type TaskType = z.infer;\nexport type Payload = z.infer;\n\nexport type EmailPayload = z.infer;\nexport type WebhookPayload = z.infer;\nexport type LogMessagePayload = z.infer;\nexport type HttpRequestPayload = z.infer;\nexport type EmailTask = z.infer;\nexport type WebhookTask = z.infer;\nexport type LogMessageTask = z.infer;\nexport type HttpRequestTask = z.infer;\n",[],[262],{"_104":270,"_95":271,"_154":272,"_156":-5,"_157":139,"_158":273,"_160":161,"_106":274,"_100":256,"_163":275,"_165":276,"_167":22,"_168":-5,"_169":277,"_98":278},"1535c40e-c8d2-11f0-86fc-42dde27851f2","worker.ts","03b2afe4-c8cd-11f0-98d4-42dde27851f2",34,["D",1763946506656],"interval","import { sqlite } from \"https://esm.town/v/std/sqlite\";\nimport { processTask } from \"./processors.ts\";\n\n/**\n * Queue worker - runs every minute to process pending tasks\n * This is an interval val that automatically runs on a schedule\n */\nexport default async function () {\n\tconsole.log(\"Queue worker starting...\");\n\n\ttry {\n\t\t// Fetch pending tasks that are ready to process\n\t\tconst result = await sqlite.execute(`\n SELECT id, task_type, payload, status, attempts, max_attempts\n FROM queue \n WHERE status = 'pending' \n AND attempts < max_attempts\n AND (scheduled_for IS NULL OR scheduled_for <= unixepoch())\n ORDER BY priority DESC, created_at ASC \n LIMIT 10\n `);\n\n\t\tconst tasks = result.rows;\n\n\t\tif (tasks.length === 0) {\n\t\t\tconsole.log(\"No tasks to process\");\n\t\t\treturn { processed: 0, success: 0, failed: 0 };\n\t\t}\n\n\t\tconsole.log(`Found ${tasks.length} tasks to process`);\n\n\t\tlet successCount = 0;\n\t\tlet failCount = 0;\n\n\t\tfor (const row of tasks) {\n\t\t\tconst task = {\n\t\t\t\tid: Number(row[0]),\n\t\t\t\ttask_type: String(row[1]),\n\t\t\t\tpayload: JSON.parse(String(row[2])),\n\t\t\t\tstatus: String(row[3]),\n\t\t\t\tattempts: Number(row[4]),\n\t\t\t\tmax_attempts: Number(row[5]),\n\t\t\t};\n\n\t\t\t// Mark as processing\n\t\t\tawait sqlite.execute({\n\t\t\t\tsql: `UPDATE queue \n SET status = 'processing', \n started_at = unixepoch(), \n attempts = attempts + 1 \n WHERE id = ?`,\n\t\t\t\targs: [task.id],\n\t\t\t});\n\n\t\t\ttry {\n\t\t\t\t// Process the task\n\t\t\t\tawait processTask(task);\n\n\t\t\t\t// Mark as completed\n\t\t\t\tawait sqlite.execute({\n\t\t\t\t\tsql: `UPDATE queue \n SET status = 'completed', \n completed_at = unixepoch() \n WHERE id = ?`,\n\t\t\t\t\targs: [task.id],\n\t\t\t\t});\n\n\t\t\t\tconsole.log(`Task ${task.id} completed successfully`);\n\t\t\t\tsuccessCount++;\n\t\t\t} catch (error) {\n\t\t\t\tconsole.error(`Task ${task.id} failed:`, error.message);\n\n\t\t\t\t// Check if max attempts reached\n\t\t\t\tconst newAttempts = task.attempts + 1;\n\t\t\t\tconst maxReached = newAttempts >= task.max_attempts;\n\n\t\t\t\tif (maxReached) {\n\t\t\t\t\t// Move to dead letter queue\n\t\t\t\t\tawait sqlite.batch([\n\t\t\t\t\t\t{\n\t\t\t\t\t\t\tsql: `INSERT INTO dead_letter_queue \n SELECT id, task_type, payload, status, created_at, started_at, \n completed_at, attempts, max_attempts, priority, ?, unixepoch()\n FROM queue WHERE id = ?`,\n\t\t\t\t\t\t\targs: [error.message, task.id],\n\t\t\t\t\t\t},\n\t\t\t\t\t\t{\n\t\t\t\t\t\t\tsql: `DELETE FROM queue WHERE id = ?`,\n\t\t\t\t\t\t\targs: [task.id],\n\t\t\t\t\t\t},\n\t\t\t\t\t]);\n\n\t\t\t\t\tconsole.log(`Task ${task.id} moved to dead letter queue`);\n\t\t\t\t} else {\n\t\t\t\t\t// Mark as failed (will retry on next run)\n\t\t\t\t\tawait sqlite.execute({\n\t\t\t\t\t\tsql: `UPDATE queue \n SET status = 'pending', \n error = ? \n WHERE id = ?`,\n\t\t\t\t\t\targs: [error.message, task.id],\n\t\t\t\t\t});\n\t\t\t\t}\n\n\t\t\t\tfailCount++;\n\t\t\t}\n\t\t}\n\n\t\tconst summary = {\n\t\t\tprocessed: tasks.length,\n\t\t\tsuccess: successCount,\n\t\t\tfailed: failCount,\n\t\t};\n\n\t\tconsole.log(\"Worker completed:\", summary);\n\t\treturn summary;\n\t} catch (error) {\n\t\tconsole.error(\"Worker error:\", error);\n\t\tthrow error;\n\t}\n}\n",[],[271],"files",[151,172,182,192,202,212,221,231,241,250,260,269],"counts",{"_283":32,"_284":178,"_285":32,"_286":287},"pulls","branches","forks","envVars",{"_288":32,"_289":32},"count","emptyCount","contents",[],{"_104":203,"_95":99,"_154":204,"_156":-5,"_157":139,"_158":101,"_160":22,"_106":293,"_100":206,"_163":207,"_165":208,"_167":22,"_168":209,"_98":294},["D",1763956437888],[99],"upstreamProject","upstreamBranch","emptyEnvVars","isNew","flags",[],"routes/_app.x.$handle.$name.branch.$branch.version.$version.code.$",{"_93":94,"_95":96,"_97":43,"_100":101,"_98":99,"_102":303,"_149":308,"_279":345,"_164":346,"_290":348,"_349":-5,"_350":-5,"_351":-5,"_352":-5,"_353":354},{"_104":105,"_95":96,"_106":304,"_108":-5,"_109":22,"_110":111,"_112":113,"_114":-5,"_14":115,"_116":-5,"_117":22,"_118":-5,"_119":32,"_16":120,"_121":122,"_123":122,"_124":125,"_126":125,"_127":125,"_128":122,"_129":122,"_23":305,"_133":306,"_136":307,"_148":22},["D",1763944165015],{"_93":94,"_131":132},{"_93":94,"_135":132,"_14":115},{"_100":138,"_95":43,"_104":139,"_140":105,"_141":-5,"_142":-5,"_143":-5,"_144":-5,"_145":-5,"_146":96,"_147":94},[309,312,315,318,321,324,327,330,333,336,339,342],{"_104":152,"_95":153,"_154":155,"_156":-5,"_157":139,"_158":159,"_160":161,"_106":162,"_100":32,"_163":164,"_165":166,"_167":22,"_168":-5,"_169":310,"_98":311},[],[153],{"_104":173,"_95":174,"_154":175,"_156":-5,"_157":139,"_158":176,"_160":161,"_106":177,"_100":178,"_163":164,"_165":179,"_167":22,"_168":-5,"_169":313,"_98":314},[],[174],{"_104":183,"_95":184,"_154":185,"_156":-5,"_157":139,"_158":186,"_160":161,"_106":187,"_100":188,"_163":164,"_165":189,"_167":22,"_168":-5,"_169":316,"_98":317},[],[184],{"_104":193,"_95":194,"_154":195,"_156":-5,"_157":139,"_158":196,"_160":161,"_106":197,"_100":198,"_163":164,"_165":199,"_167":22,"_168":-5,"_169":319,"_98":320},[],[194],{"_104":203,"_95":99,"_154":204,"_156":-5,"_157":139,"_158":101,"_160":22,"_106":205,"_100":206,"_163":207,"_165":208,"_167":22,"_168":209,"_169":322,"_98":323},[],[99],{"_104":213,"_95":214,"_154":215,"_156":-5,"_157":139,"_158":216,"_160":161,"_106":217,"_100":32,"_163":164,"_165":218,"_167":22,"_168":-5,"_169":325,"_98":326},[],[214],{"_104":222,"_95":223,"_154":224,"_156":-5,"_157":139,"_158":225,"_160":161,"_106":226,"_100":227,"_163":164,"_165":228,"_167":22,"_168":-5,"_169":328,"_98":329},[],[223],{"_104":232,"_95":233,"_154":234,"_156":-5,"_157":139,"_158":235,"_160":161,"_106":236,"_100":198,"_163":237,"_165":238,"_167":22,"_168":-5,"_169":331,"_98":332},[],[233],{"_104":242,"_95":243,"_154":244,"_156":-5,"_157":139,"_158":245,"_160":161,"_106":246,"_100":198,"_163":207,"_165":247,"_167":22,"_168":-5,"_169":334,"_98":335},[],[243],{"_104":251,"_95":252,"_154":253,"_156":-5,"_157":139,"_158":254,"_160":161,"_106":255,"_100":256,"_163":237,"_165":257,"_167":22,"_168":-5,"_169":337,"_98":338},[],[252],{"_104":261,"_95":262,"_154":263,"_156":-5,"_157":139,"_158":264,"_160":161,"_106":265,"_100":178,"_163":237,"_165":266,"_167":22,"_168":-5,"_169":340,"_98":341},[],[262],{"_104":270,"_95":271,"_154":272,"_156":-5,"_157":139,"_158":273,"_160":161,"_106":274,"_100":256,"_163":275,"_165":276,"_167":22,"_168":-5,"_169":343,"_98":344},[],[271],[309,312,315,318,321,324,327,330,333,336,339,342],{"_104":203,"_95":99,"_154":204,"_156":-5,"_157":139,"_158":101,"_160":22,"_106":293,"_100":206,"_163":207,"_165":208,"_167":22,"_168":209,"_98":347},[99],[],"upstream","pull","ahead","behind","templates",[],"actionData","errors"]

nbbaier

lightweightQueue