feat(redis): integrate Redis for job queue management and persistent storage
This commit is contained in:
MythEclipse
@@ -165,6 +165,7 @@ const configSchema = z
|
||||
POSTGRES_POOL_MIN: z.coerce.number().int().positive().default(2),
|
||||
POSTGRES_POOL_MAX: z.coerce.number().int().positive().default(10),
|
||||
ADMIN_PASSWORD: z.string().default("admin123"),
|
||||
REDIS_URL: z.string().min(1).default("redis://localhost:6379"),
|
||||
})
|
||||
.superRefine((value, ctx) => {
|
||||
if (!value.AI_ANALYSIS_ENABLED) {
|
||||
|
||||
+289
-216
@@ -1,35 +1,37 @@
|
||||
import { and, asc, eq, lt, sql } from "drizzle-orm";
|
||||
import {
|
||||
getDatabase as getDrizzleDatabase,
|
||||
initializeDatabase,
|
||||
} from "./database/drizzle.js";
|
||||
import { muxerJobsTable, uiStateTable } from "./database/schema.js";
|
||||
import Redis from "ioredis";
|
||||
import { config } from "./config.js";
|
||||
import { createChildLogger } from "./logger.js";
|
||||
|
||||
const logger = createChildLogger("muxer-queue");
|
||||
|
||||
interface QueryBuilder<T = unknown> extends PromiseLike<T> {
|
||||
from(...args: unknown[]): QueryBuilder<T>;
|
||||
where(...args: unknown[]): QueryBuilder<T>;
|
||||
orderBy(...args: unknown[]): QueryBuilder<T>;
|
||||
limit(...args: unknown[]): QueryBuilder<T>;
|
||||
values(...args: unknown[]): QueryBuilder<T>;
|
||||
onConflictDoNothing(...args: unknown[]): QueryBuilder<T>;
|
||||
onConflictDoUpdate(...args: unknown[]): QueryBuilder<T>;
|
||||
set(...args: unknown[]): QueryBuilder<T>;
|
||||
groupBy(...args: unknown[]): QueryBuilder<T>;
|
||||
// ── Redis client (lazy singleton) ──────────────────────────────────────────
|
||||
|
||||
let redis: Redis | null = null;
|
||||
|
||||
function getRedis(): Redis {
|
||||
if (redis !== null) return redis;
|
||||
|
||||
redis = new Redis(config.REDIS_URL, {
|
||||
maxRetriesPerRequest: 3,
|
||||
retryStrategy(times) {
|
||||
if (times > 5) return null; // stop retrying
|
||||
return Math.min(times * 200, 2000);
|
||||
},
|
||||
lazyConnect: false,
|
||||
});
|
||||
|
||||
redis.on("error", (err) => {
|
||||
logger.error({ err }, "Redis connection error");
|
||||
});
|
||||
|
||||
redis.on("connect", () => {
|
||||
logger.info({ url: config.REDIS_URL }, "Redis connected");
|
||||
});
|
||||
|
||||
return redis;
|
||||
}
|
||||
|
||||
export interface SqliteDatabase {
|
||||
select<T = unknown[]>(...args: unknown[]): QueryBuilder<T>;
|
||||
insert(...args: unknown[]): QueryBuilder<unknown>;
|
||||
update(...args: unknown[]): QueryBuilder<unknown>;
|
||||
delete(...args: unknown[]): QueryBuilder<unknown>;
|
||||
}
|
||||
|
||||
function db(): SqliteDatabase {
|
||||
return getDrizzleDatabase() as unknown as SqliteDatabase;
|
||||
}
|
||||
// ── Types ──────────────────────────────────────────────────────────────────
|
||||
|
||||
export interface MuxerJobData {
|
||||
userId: string;
|
||||
@@ -38,56 +40,37 @@ export interface MuxerJobData {
|
||||
outputDir: string;
|
||||
}
|
||||
|
||||
interface StoredJobRow {
|
||||
id: string;
|
||||
data: string;
|
||||
status: "pending" | "processing" | "completed" | "failed";
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
error: string | null;
|
||||
// ── Database backward compatibility ────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* @deprecated Use Redis functions directly. Kept for backward compat with old
|
||||
* tests that import getDatabase from muxer-queue.
|
||||
*/
|
||||
export function getDatabase() {
|
||||
logger.warn(
|
||||
"getDatabase() is deprecated — queue now uses Redis. Returning a stub.",
|
||||
);
|
||||
return undefined as unknown as never;
|
||||
}
|
||||
|
||||
interface JobStatsRow {
|
||||
status: "pending" | "processing" | "completed" | "failed";
|
||||
count: number | string | { count: number | string };
|
||||
}
|
||||
// ── Persistent KV store (replaces SQLite uiState table) ────────────────────
|
||||
|
||||
interface StoredJob {
|
||||
id: string;
|
||||
data: string;
|
||||
status: "pending" | "processing" | "completed" | "failed";
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
// Export getDatabase for backward compatibility with webserver.ts
|
||||
export function getDatabase(): SqliteDatabase {
|
||||
return db();
|
||||
}
|
||||
const KV_PREFIX = "kv:";
|
||||
|
||||
export async function getPersistedValue<T>(
|
||||
key: string,
|
||||
fallback: T,
|
||||
): Promise<T> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
|
||||
const row = await database
|
||||
.select<Array<{ value: string }>>()
|
||||
.from(uiStateTable)
|
||||
.where(eq(uiStateTable.key, key))
|
||||
.limit(1);
|
||||
|
||||
if (!row || row.length === 0) return fallback;
|
||||
|
||||
try {
|
||||
return JSON.parse(row[0].value) as T;
|
||||
} catch {
|
||||
const r = getRedis();
|
||||
const raw = await r.get(`${KV_PREFIX}${key}`);
|
||||
if (raw === null) return fallback;
|
||||
return JSON.parse(raw) as T;
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
{ key, error: error instanceof Error ? error.message : String(error) },
|
||||
"Failed to get persisted value",
|
||||
);
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
@@ -96,45 +79,62 @@ export async function setPersistedValue(
|
||||
key: string,
|
||||
value: unknown,
|
||||
): Promise<void> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
try {
|
||||
const r = getRedis();
|
||||
await r.set(`${KV_PREFIX}${key}`, JSON.stringify(value));
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
{ key, error: error instanceof Error ? error.message : String(error) },
|
||||
"Failed to set persisted value",
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
await database
|
||||
.insert(uiStateTable)
|
||||
.values({
|
||||
key,
|
||||
value: JSON.stringify(value),
|
||||
updated_at: Date.now(),
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: uiStateTable.key,
|
||||
set: {
|
||||
value: JSON.stringify(value),
|
||||
updated_at: Date.now(),
|
||||
},
|
||||
});
|
||||
// ── Job queue ──────────────────────────────────────────────────────────────
|
||||
|
||||
const JOB_PREFIX = "job:";
|
||||
const QUEUE_PENDING = "queue:pending";
|
||||
const QUEUE_PROCESSING = "queue:processing";
|
||||
const QUEUE_COMPLETED = "queue:completed";
|
||||
const QUEUE_FAILED = "queue:failed";
|
||||
|
||||
function queueKey(status: string): string {
|
||||
switch (status) {
|
||||
case "pending":
|
||||
return QUEUE_PENDING;
|
||||
case "processing":
|
||||
return QUEUE_PROCESSING;
|
||||
case "completed":
|
||||
return QUEUE_COMPLETED;
|
||||
case "failed":
|
||||
return QUEUE_FAILED;
|
||||
default:
|
||||
return QUEUE_PENDING;
|
||||
}
|
||||
}
|
||||
|
||||
export async function enqueueMuxerJob(data: MuxerJobData): Promise<string> {
|
||||
try {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
|
||||
const r = getRedis();
|
||||
const jobId = `${data.userId}-${data.sessionId}`;
|
||||
const now = Date.now();
|
||||
|
||||
await database
|
||||
.insert(muxerJobsTable)
|
||||
.values({
|
||||
id: jobId,
|
||||
data: JSON.stringify(data),
|
||||
status: "pending",
|
||||
attempts: 0,
|
||||
maxAttempts: 3,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
const jobKey = `${JOB_PREFIX}${jobId}`;
|
||||
|
||||
// Use a pipeline for atomicity
|
||||
const pipeline = r.pipeline();
|
||||
pipeline.hset(jobKey, {
|
||||
data: JSON.stringify(data),
|
||||
status: "pending",
|
||||
attempts: "0",
|
||||
maxAttempts: "3",
|
||||
createdAt: String(now),
|
||||
updatedAt: String(now),
|
||||
error: "",
|
||||
});
|
||||
pipeline.lpush(QUEUE_PENDING, jobId);
|
||||
await pipeline.exec();
|
||||
|
||||
logger.info(
|
||||
{ jobId, userId: data.userId, sessionId: data.sessionId },
|
||||
@@ -154,27 +154,69 @@ export async function enqueueMuxerJob(data: MuxerJobData): Promise<string> {
|
||||
}
|
||||
}
|
||||
|
||||
export async function getPendingJobs(): Promise<StoredJob[]> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
export async function getPendingJobs(): Promise<
|
||||
Array<{
|
||||
id: string;
|
||||
data: string;
|
||||
status: "pending" | "processing" | "completed" | "failed";
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
error?: string;
|
||||
}>
|
||||
> {
|
||||
try {
|
||||
const r = getRedis();
|
||||
|
||||
const rows = await database
|
||||
.select<StoredJobRow[]>()
|
||||
.from(muxerJobsTable)
|
||||
.where(eq(muxerJobsTable.status, "pending"))
|
||||
.orderBy(asc(muxerJobsTable.createdAt))
|
||||
.limit(10);
|
||||
// Get up to 10 pending job IDs from the left (oldest first)
|
||||
const jobIds = await r.lrange(QUEUE_PENDING, 0, 9);
|
||||
if (jobIds.length === 0) return [];
|
||||
|
||||
return rows.map((row) => ({
|
||||
id: row.id,
|
||||
data: row.data,
|
||||
status: row.status as "pending" | "processing" | "completed" | "failed",
|
||||
attempts: row.attempts,
|
||||
maxAttempts: row.maxAttempts,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
error: row.error || undefined,
|
||||
}));
|
||||
// Batch fetch all job hashes
|
||||
const pipeline = r.pipeline();
|
||||
for (const id of jobIds) {
|
||||
pipeline.hgetall(`${JOB_PREFIX}${id}`);
|
||||
}
|
||||
const results = await pipeline.exec();
|
||||
if (!results) return [];
|
||||
|
||||
const jobs: Array<{
|
||||
id: string;
|
||||
data: string;
|
||||
status: "pending" | "processing" | "completed" | "failed";
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
error?: string;
|
||||
}> = [];
|
||||
|
||||
for (let i = 0; i < jobIds.length; i++) {
|
||||
const [err, fields] = results[i];
|
||||
if (err || !fields) continue;
|
||||
|
||||
const raw = fields as Record<string, string>;
|
||||
jobs.push({
|
||||
id: jobIds[i],
|
||||
data: raw.data || "",
|
||||
status: (raw.status as "pending") || "pending",
|
||||
attempts: Number(raw.attempts) || 0,
|
||||
maxAttempts: Number(raw.maxAttempts) || 3,
|
||||
createdAt: Number(raw.createdAt) || 0,
|
||||
updatedAt: Number(raw.updatedAt) || 0,
|
||||
error: raw.error || undefined,
|
||||
});
|
||||
}
|
||||
|
||||
return jobs;
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
{ error: error instanceof Error ? error.message : String(error) },
|
||||
"Failed to get pending jobs",
|
||||
);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
export async function updateJobStatus(
|
||||
@@ -182,95 +224,135 @@ export async function updateJobStatus(
|
||||
status: "processing" | "completed" | "failed",
|
||||
error?: string,
|
||||
): Promise<void> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
const now = Date.now();
|
||||
try {
|
||||
const r = getRedis();
|
||||
const jobKey = `${JOB_PREFIX}${jobId}`;
|
||||
const now = Date.now();
|
||||
|
||||
if (status === "failed") {
|
||||
await database
|
||||
.update(muxerJobsTable)
|
||||
.set({
|
||||
status,
|
||||
attempts: sql`${muxerJobsTable.attempts} + 1`,
|
||||
updatedAt: now,
|
||||
error: error || null,
|
||||
})
|
||||
.where(eq(muxerJobsTable.id, jobId));
|
||||
} else {
|
||||
await database
|
||||
.update(muxerJobsTable)
|
||||
.set({
|
||||
status,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(muxerJobsTable.id, jobId));
|
||||
const exists = await r.exists(jobKey);
|
||||
if (!exists) {
|
||||
logger.warn({ jobId }, "Job not found for status update");
|
||||
return;
|
||||
}
|
||||
|
||||
const currentStatus = await r.hget(jobKey, "status");
|
||||
|
||||
const pipeline = r.pipeline();
|
||||
|
||||
if (status === "failed") {
|
||||
pipeline.hincrby(jobKey, "attempts", 1);
|
||||
pipeline.hset(jobKey, "error", error || "");
|
||||
}
|
||||
|
||||
pipeline.hset(jobKey, "status", status);
|
||||
pipeline.hset(jobKey, "updatedAt", String(now));
|
||||
|
||||
// Move job ID between queue lists
|
||||
if (currentStatus) {
|
||||
pipeline.lrem(queueKey(currentStatus), 0, jobId);
|
||||
}
|
||||
pipeline.lpush(queueKey(status), jobId);
|
||||
|
||||
await pipeline.exec();
|
||||
|
||||
logger.info({ jobId, status, error }, "Job status updated");
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
{
|
||||
jobId,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
},
|
||||
"Failed to update job status",
|
||||
);
|
||||
throw err;
|
||||
}
|
||||
|
||||
logger.info({ jobId, status, error }, "Job status updated");
|
||||
}
|
||||
|
||||
export async function retryFailedJob(jobId: string): Promise<boolean> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
try {
|
||||
const r = getRedis();
|
||||
const jobKey = `${JOB_PREFIX}${jobId}`;
|
||||
|
||||
const jobs = await database
|
||||
.select<StoredJobRow[]>()
|
||||
.from(muxerJobsTable)
|
||||
.where(eq(muxerJobsTable.id, jobId))
|
||||
.limit(1);
|
||||
const [attemptsStr, maxAttemptsStr] = await r.hmget(
|
||||
jobKey,
|
||||
"attempts",
|
||||
"maxAttempts",
|
||||
);
|
||||
|
||||
const job = jobs[0];
|
||||
const attempts = Number(attemptsStr) || 0;
|
||||
const maxAttempts = Number(maxAttemptsStr) || 3;
|
||||
|
||||
if (!job) {
|
||||
logger.warn({ jobId }, "Job not found");
|
||||
return false;
|
||||
}
|
||||
if (attempts >= maxAttempts) {
|
||||
logger.warn(
|
||||
{ jobId, attempts, maxAttempts },
|
||||
"Max retry attempts reached",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (job.attempts >= job.maxAttempts) {
|
||||
logger.warn(
|
||||
{ jobId, attempts: job.attempts, maxAttempts: job.maxAttempts },
|
||||
"Max retry attempts reached",
|
||||
const pipeline = r.pipeline();
|
||||
pipeline.hset(jobKey, "status", "pending");
|
||||
pipeline.hset(jobKey, "updatedAt", String(Date.now()));
|
||||
pipeline.lrem(QUEUE_FAILED, 0, jobId);
|
||||
pipeline.lpush(QUEUE_PENDING, jobId);
|
||||
await pipeline.exec();
|
||||
|
||||
logger.info({ jobId, attempt: attempts + 1 }, "Job retried");
|
||||
return true;
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
{ jobId, error: err instanceof Error ? err.message : String(err) },
|
||||
"Failed to retry job",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
await database
|
||||
.update(muxerJobsTable)
|
||||
.set({
|
||||
status: "pending",
|
||||
updatedAt: Date.now(),
|
||||
})
|
||||
.where(eq(muxerJobsTable.id, jobId));
|
||||
|
||||
logger.info({ jobId, attempt: job.attempts + 1 }, "Job retried");
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
export async function cleanupCompletedJobs(
|
||||
olderThanMs: number = 24 * 60 * 60 * 1000,
|
||||
): Promise<number> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
const cutoffTime = Date.now() - olderThanMs;
|
||||
try {
|
||||
const r = getRedis();
|
||||
const cutoffTime = Date.now() - olderThanMs;
|
||||
|
||||
const result = await database
|
||||
.delete(muxerJobsTable)
|
||||
.where(
|
||||
and(
|
||||
eq(muxerJobsTable.status, "completed"),
|
||||
lt(muxerJobsTable.updatedAt, cutoffTime),
|
||||
),
|
||||
const jobIds = await r.lrange(QUEUE_COMPLETED, 0, -1);
|
||||
if (jobIds.length === 0) return 0;
|
||||
|
||||
const pipeline = r.pipeline();
|
||||
for (const id of jobIds) {
|
||||
pipeline.hget(`${JOB_PREFIX}${id}`, "updatedAt");
|
||||
}
|
||||
const results = await pipeline.exec();
|
||||
if (!results) return 0;
|
||||
|
||||
let deletedCount = 0;
|
||||
const deletePipeline = r.pipeline();
|
||||
|
||||
for (let i = 0; i < jobIds.length; i++) {
|
||||
const [err, updatedAtStr] = results[i];
|
||||
if (err) continue;
|
||||
|
||||
const updatedAt = Number(updatedAtStr) || 0;
|
||||
if (updatedAt < cutoffTime) {
|
||||
deletePipeline.del(`${JOB_PREFIX}${jobIds[i]}`);
|
||||
deletePipeline.lrem(QUEUE_COMPLETED, 0, jobIds[i]);
|
||||
deletedCount++;
|
||||
}
|
||||
}
|
||||
|
||||
if (deletedCount > 0) {
|
||||
await deletePipeline.exec();
|
||||
}
|
||||
|
||||
logger.info({ deletedCount }, "Cleaned up completed jobs");
|
||||
return deletedCount;
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
{ error: err instanceof Error ? err.message : String(err) },
|
||||
"Failed to clean up completed jobs",
|
||||
);
|
||||
|
||||
const deletedCount =
|
||||
typeof result === "object" && result !== null && "rowsAffected" in result
|
||||
? Number(result.rowsAffected)
|
||||
: 0;
|
||||
|
||||
logger.info({ deletedCount }, "Cleaned up completed jobs");
|
||||
|
||||
return deletedCount;
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
export async function getJobStats(): Promise<{
|
||||
@@ -279,38 +361,29 @@ export async function getJobStats(): Promise<{
|
||||
completed: number;
|
||||
failed: number;
|
||||
}> {
|
||||
await initializeDatabase();
|
||||
const database = db();
|
||||
try {
|
||||
const r = getRedis();
|
||||
const [pending, processing, completed, failed] = await Promise.all([
|
||||
r.llen(QUEUE_PENDING),
|
||||
r.llen(QUEUE_PROCESSING),
|
||||
r.llen(QUEUE_COMPLETED),
|
||||
r.llen(QUEUE_FAILED),
|
||||
]);
|
||||
|
||||
const rows = await database
|
||||
.select<JobStatsRow[]>({
|
||||
status: muxerJobsTable.status,
|
||||
count: sql<number>`COUNT(*)`,
|
||||
})
|
||||
.from(muxerJobsTable)
|
||||
.groupBy(muxerJobsTable.status);
|
||||
|
||||
const stats = {
|
||||
pending: 0,
|
||||
processing: 0,
|
||||
completed: 0,
|
||||
failed: 0,
|
||||
};
|
||||
|
||||
for (const row of rows) {
|
||||
const count =
|
||||
typeof row.count === "object" && "count" in row.count
|
||||
? Number((row.count as { count: number | string }).count)
|
||||
: Number(row.count);
|
||||
if (row.status === "pending") stats.pending = count;
|
||||
else if (row.status === "processing") stats.processing = count;
|
||||
else if (row.status === "completed") stats.completed = count;
|
||||
else if (row.status === "failed") stats.failed = count;
|
||||
return { pending, processing, completed, failed };
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
{ error: err instanceof Error ? err.message : String(err) },
|
||||
"Failed to get job stats",
|
||||
);
|
||||
return { pending: 0, processing: 0, completed: 0, failed: 0 };
|
||||
}
|
||||
|
||||
return stats;
|
||||
}
|
||||
|
||||
export async function closeQueue(): Promise<void> {
|
||||
logger.info("Muxer queue closed");
|
||||
if (redis !== null) {
|
||||
await redis.quit();
|
||||
redis = null;
|
||||
}
|
||||
logger.info("Muxer queue (Redis) closed");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user