feat: add BullMQ ingest queue and processor
Idempotent message persistence via prisma.message.upsert. Uses URL parsing to pass connection options to BullMQ, avoiding ioredis version conflicts. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,11 @@
|
||||
import { Worker } from 'bullmq';
|
||||
import { IngestJobData } from '@tower/types';
|
||||
|
||||
function parseRedisUrl(url: string) {
|
||||
const { hostname, port } = new URL(url);
|
||||
return { host: hostname, port: parseInt(port || '6379', 10), maxRetriesPerRequest: null };
|
||||
}
|
||||
|
||||
export async function processIngestJob(job: IngestJobData, prisma: any): Promise<void> {
|
||||
await prisma.message.upsert({
|
||||
where: {
|
||||
@@ -24,10 +29,9 @@ export async function processIngestJob(job: IngestJobData, prisma: any): Promise
|
||||
}
|
||||
|
||||
export function createIngestWorker(redisUrl: string, prisma: any): Worker<IngestJobData> {
|
||||
const connection = { host: 'localhost', port: 6379, maxRetriesPerRequest: null } as any;
|
||||
return new Worker<IngestJobData>(
|
||||
'tower:ingest',
|
||||
async (job) => processIngestJob(job.data, prisma),
|
||||
{ connection },
|
||||
{ connection: parseRedisUrl(redisUrl) },
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user