added email

.env.example

@@ -42,5 +42,10 @@ LLM_REASONING_EFFORT=low
 CORPUS_CACHE_DIR=.corpus-cache
 RESEARCH_MAX_STEPS=10
 
+# ---- Worker (jobs queue lives in Postgres; no Redis)
+WORKER_CONCURRENCY=4
+WORKER_POLL_SECONDS=1
+JOB_STALE_SECONDS=900
+
 # ---- ElevenLabs (Phase 1 audio)
 ELEVENLABS_API_KEY=

.github/workflows/ci.yml

@@ -5,6 +5,14 @@ jobs:
   backend:
     runs-on: ubuntu-latest
     defaults: { run: { working-directory: backend } }
+    services:
+      postgres:
+        image: postgres:16-alpine
+        env: { POSTGRES_USER: notestack, POSTGRES_PASSWORD: notestack, POSTGRES_DB: notestack_test }
+        ports: ["5432:5432"]
+        options: --health-cmd "pg_isready -U notestack" --health-interval 5s --health-retries 10
+    env:
+      TEST_POSTGRES_URL: postgresql+psycopg://notestack:notestack@localhost:5432/notestack_test
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5

README.md

@@ -9,7 +9,7 @@ overviews, videos and launch kits, all traced back to what the writer actually w
 ## Layout
 
 ```
-backend/    FastAPI + SQLAlchemy + Alembic, arq worker, DSPy (LLM agnostic, Z.ai first)
+backend/    FastAPI + SQLAlchemy + Alembic, Postgres job queue worker, DSPy (LLM agnostic, Z.ai first)
 frontend/   React + Vite + TS: night sky landing, pricing, blog, auth, Mission Control
 renderer/   Remotion compositions + render service that uploads to R2 via presigned PUT
 docs/       Design doc and task lists
@@ -31,8 +31,8 @@ Without Docker:
 
 ```bash
 cd backend && python -m venv .venv && .venv/Scripts/pip install -r requirements-dev.txt
-alembic upgrade head && uvicorn app.main:app --reload     # needs Postgres + Redis
-arq app.worker.WorkerSettings                              # second terminal
+alembic upgrade head && uvicorn app.main:app --reload     # needs Postgres only
+python -m app.worker                                       # second terminal
 cd frontend && npm install && npm run dev
 cd renderer && npm install && npm start                    # or: npm run studio
 ```

backend/app/config.py

@@ -3,7 +3,6 @@ from pathlib import Path
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-
 # One .env at the repo root serves every way of running the API; a backend/.env (or the
 # working directory's .env) overrides it. Real environment variables beat both.
 _ROOT_ENV = Path(__file__).resolve().parents[2] / ".env"
@@ -19,7 +18,11 @@ class Settings(BaseSettings):
     cors_origins: str = "http://localhost:5173"
 
     database_url: str = "postgresql+psycopg://notestack:notestack@localhost:5432/notestack"
-    redis_url: str = "redis://localhost:6379/0"
+
+    # Worker (Postgres backed queue, see app/worker.py)
+    worker_concurrency: int = 4
+    worker_poll_seconds: float = 1.0
+    job_stale_seconds: int = 900  # a running job with no heartbeat for this long is retried
 
     # Auth
     jwt_secret: str = "change-me"

backend/app/main.py

@@ -3,8 +3,6 @@ import time
 import uuid
 from contextlib import asynccontextmanager
 
-from arq import create_pool
-from arq.connections import RedisSettings
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 
@@ -19,7 +17,6 @@ log = logging.getLogger("notestack")
 async def lifespan(app: FastAPI):
     logging.basicConfig(level=logging.INFO)
     configure_default()
-    app.state.arq = await create_pool(RedisSettings.from_dsn(settings.redis_url))
     if settings.env == "development" and settings.r2_endpoint_url:
         try:
             from app.services.storage import storage as store
@@ -28,7 +25,6 @@ async def lifespan(app: FastAPI):
         except Exception:
             log.warning("could not ensure local bucket", exc_info=True)
     yield
-    await app.state.arq.aclose()
 
 
 app = FastAPI(title="Notestack API", version="0.1.0", lifespan=lifespan)

backend/app/models/content.py

@@ -8,6 +8,7 @@ from sqlalchemy import (
     DateTime,
     Float,
     ForeignKey,
+    Index,
     Integer,
     String,
     Text,
@@ -142,9 +143,14 @@ class Artifact(IdMixin, TimestampMixin, Base):
 
 
 class Job(IdMixin, TimestampMixin, Base):
-    """Covers ingest, tts and render jobs. kind decides which fields matter."""
+    """Covers ingest, tts and render jobs. kind decides which fields matter.
+
+    This table is also the queue: workers claim queued rows with FOR UPDATE SKIP LOCKED
+    (see app/worker.py), so no separate broker is needed.
+    """
 
     __tablename__ = "jobs"
+    __table_args__ = (Index("ix_jobs_queue", "status", "run_after", "created_at"),)
 
     workspace_id: Mapped[uuid.UUID] = _ws_fk()
     kind: Mapped[str] = mapped_column(String(30))  # ingest | tts | render | generate
@@ -158,6 +164,13 @@ class Job(IdMixin, TimestampMixin, Base):
     cost_usd: Mapped[float] = mapped_column(Float, default=0.0)
     finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
 
+    # Queue bookkeeping
+    attempts: Mapped[int] = mapped_column(Integer, default=0)
+    max_attempts: Mapped[int] = mapped_column(Integer, default=3)
+    run_after: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # retry backoff
+    locked_by: Mapped[str | None] = mapped_column(String(100))  # worker id while running
+    heartbeat_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # stale detection
+
 
 class Upload(IdMixin, TimestampMixin, Base):
     __tablename__ = "uploads"

backend/app/routers/artifacts.py

@@ -1,7 +1,7 @@
 import uuid
 from typing import Literal
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy import select
 
@@ -66,12 +66,15 @@ def get_artifact(artifact_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
 
 
 @router.post("/{artifact_id}/render")
-async def render_artifact(artifact_id: uuid.UUID, body: RenderIn, request: Request, ctx: Ctx = Depends(get_ctx)):
+def render_artifact(artifact_id: uuid.UUID, body: RenderIn, ctx: Ctx = Depends(get_ctx)):
     a = _get(ctx, artifact_id)
-    fmt = COMPOSITIONS[body.composition]
-    job = create_job(ctx.db, ctx.workspace.id, "render", {"composition": body.composition})
-    job.artifact_id = a.id
     a.status = "rendering"
-    ctx.db.commit()
-    await request.app.state.arq.enqueue_job("render_task", str(a.id), str(job.id), body.composition, body.props, fmt)
+    job = create_job(
+        ctx.db,
+        ctx.workspace.id,
+        "render",
+        {"artifact_id": str(a.id), "composition": body.composition, "props": body.props,
+         "format": COMPOSITIONS[body.composition]},
+        artifact_id=a.id,
+    )
     return serialize_job(job)

backend/app/routers/jobs.py

@@ -1,4 +1,3 @@
-import json
 import uuid
 
 from fastapi import APIRouter, Depends, Header, HTTPException
@@ -9,9 +8,9 @@ from sqlalchemy.orm import Session
 
 from app.auth import Ctx, get_ctx
 from app.config import settings
-from app.db import get_db
+from app.db import SessionLocal, get_db
 from app.models import Artifact, Job
-from app.services.jobs import record_usage, serialize_job, subscribe, update_job
+from app.services.jobs import record_usage, serialize_job, update_job, watch_job
 
 router = APIRouter(prefix="/api", tags=["jobs"])
 
@@ -30,17 +29,11 @@ def get_job(job_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
 
 @router.get("/jobs/{job_id}/events")
 async def job_events(job_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
-    job = _get(ctx, job_id)
-    snapshot = serialize_job(job)
+    _get(ctx, job_id)  # ownership check
 
     async def stream():
-        yield f"data: {json.dumps(snapshot)}\n\n"
-        if snapshot["status"] in {"done", "failed"}:
-            return
-        async for data in subscribe(job_id):
+        async for data in watch_job(SessionLocal, job_id):
             yield f"data: {data}\n\n"
-            if json.loads(data).get("status") in {"done", "failed"}:
-                return
 
     return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"})
 

backend/app/routers/sources.py

@@ -1,6 +1,6 @@
 import uuid
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy import func, select
 
@@ -30,20 +30,18 @@ def _serialize(source: Source, doc_count: int = 0) -> dict:
     }
 
 
-async def _enqueue_sync(request: Request, ctx: Ctx, source: Source) -> Job:
-    job = create_job(ctx.db, ctx.workspace.id, "ingest", {"source_id": str(source.id)})
-    await request.app.state.arq.enqueue_job("ingest_source_task", str(source.id), str(job.id))
-    return job
+def _enqueue_sync(ctx: Ctx, source: Source) -> Job:
+    return create_job(ctx.db, ctx.workspace.id, "ingest", {"source_id": str(source.id)})
 
 
 @router.post("")
-async def create_source(body: SourceIn, request: Request, ctx: Ctx = Depends(get_ctx)):
+def create_source(body: SourceIn, ctx: Ctx = Depends(get_ctx)):
     feed_url, platform = normalize_feed_url(body.url)
     existing = ctx.db.scalar(
         select(Source).where(Source.workspace_id == ctx.workspace.id, Source.feed_url == feed_url)
     )
     if existing:
-        job = await _enqueue_sync(request, ctx, existing)
+        job = _enqueue_sync(ctx, existing)
         return {"source": _serialize(existing), "job": serialize_job(job)}
     plan = effective_plan(ctx.db, ctx.workspace)
     count = ctx.db.scalar(select(func.count()).select_from(Source).where(Source.workspace_id == ctx.workspace.id))
@@ -52,7 +50,7 @@ async def create_source(body: SourceIn, request: Request, ctx: Ctx = Depends(get
     source = Source(workspace_id=ctx.workspace.id, feed_url=feed_url, platform=platform, site_url=body.url)
     ctx.db.add(source)
     ctx.db.commit()
-    job = await _enqueue_sync(request, ctx, source)
+    job = _enqueue_sync(ctx, source)
     return {"source": _serialize(source), "job": serialize_job(job)}
 
 
@@ -69,13 +67,13 @@ def list_sources(ctx: Ctx = Depends(get_ctx)):
 
 
 @router.get("/{source_id}/sync")
-async def sync_source(source_id: uuid.UUID, request: Request, ctx: Ctx = Depends(get_ctx)):
+def sync_source(source_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
     source = ctx.db.scalar(
         select(Source).where(Source.id == source_id, Source.workspace_id == ctx.workspace.id)
     )
     if not source:
         raise HTTPException(404, "Source not found")
-    job = await _enqueue_sync(request, ctx, source)
+    job = _enqueue_sync(ctx, source)
     return {"source": _serialize(source), "job": serialize_job(job)}
 
 

backend/app/services/jobs.py

@@ -1,28 +1,28 @@
-"""Job rows + progress fan-out over Redis pub/sub (consumed by the SSE endpoint)."""
+"""Jobs on Postgres: the jobs table is the queue, the progress record and the SSE source.
 
+Enqueue = insert a queued row. Workers claim with FOR UPDATE SKIP LOCKED, so any number of worker
+processes can poll the same table without double-running a job. Progress lives on the row, and the
+SSE endpoint polls it.
+"""
+
+import asyncio
 import json
 import uuid
-from datetime import UTC, datetime
+from collections.abc import AsyncIterator, Callable
+from datetime import UTC, datetime, timedelta
 
-import redis
-import redis.asyncio as aredis
+from fastapi.concurrency import run_in_threadpool
+from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.models import Job, UsageEvent
 
-_sync_redis: redis.Redis | None = None
-
-
-def _redis() -> redis.Redis:
-    global _sync_redis
-    if _sync_redis is None:
-        _sync_redis = redis.Redis.from_url(settings.redis_url)
-    return _sync_redis
+TERMINAL = {"done", "failed"}
 
 
-def channel(job_id: uuid.UUID | str) -> str:
-    return f"job:{job_id}"
+def utcnow() -> datetime:
+    return datetime.now(UTC)
 
 
 def serialize_job(job: Job) -> dict:
@@ -35,14 +35,34 @@ def serialize_job(job: Job) -> dict:
         "error": job.error,
         "result": job.result,
         "artifact_id": str(job.artifact_id) if job.artifact_id else None,
+        "attempts": job.attempts,
         "cost_usd": job.cost_usd,
         "created_at": job.created_at.isoformat() if job.created_at else None,
         "finished_at": job.finished_at.isoformat() if job.finished_at else None,
     }
 
 
-def create_job(db: Session, workspace_id: uuid.UUID, kind: str, params: dict | None = None) -> Job:
-    job = Job(workspace_id=workspace_id, kind=kind, params=params or {})
+# Producer side
+
+
+def create_job(
+    db: Session,
+    workspace_id: uuid.UUID,
+    kind: str,
+    params: dict | None = None,
+    *,
+    artifact_id: uuid.UUID | None = None,
+    max_attempts: int = 3,
+) -> Job:
+    """Enqueue. The worker picks it up within WORKER_POLL_SECONDS."""
+    job = Job(
+        workspace_id=workspace_id,
+        kind=kind,
+        params=params or {},
+        artifact_id=artifact_id,
+        status="queued",
+        max_attempts=max_attempts,
+    )
     db.add(job)
     db.commit()
     return job
@@ -58,10 +78,12 @@ def update_job(
     error: str | None = None,
     result: dict | None = None,
 ) -> None:
+    """Every update is also a heartbeat, so a job that keeps reporting is never treated as stale."""
     if status:
         job.status = status
-        if status in {"done", "failed"}:
-            job.finished_at = datetime.now(UTC)
+        if status in TERMINAL:
+            job.finished_at = utcnow()
+            job.locked_by = None
     if progress is not None:
         job.progress = max(0.0, min(1.0, progress))
     if message is not None:
@@ -70,24 +92,92 @@ def update_job(
         job.error = error
     if result is not None:
         job.result = result
+    job.heartbeat_at = utcnow()
     db.commit()
-    try:
-        _redis().publish(channel(job.id), json.dumps(serialize_job(job)))
-    except redis.RedisError:
-        pass  # progress is also readable from GET /jobs/{id}
-
-
-async def subscribe(job_id: uuid.UUID):
-    client = aredis.Redis.from_url(settings.redis_url)
-    pubsub = client.pubsub()
-    await pubsub.subscribe(channel(job_id))
-    try:
-        async for msg in pubsub.listen():
-            if msg["type"] == "message":
-                yield msg["data"].decode()
-    finally:
-        await pubsub.unsubscribe(channel(job_id))
-        await client.aclose()
+
+
+# Consumer side (worker)
+
+
+def claim_next(db: Session, worker_id: str) -> uuid.UUID | None:
+    """Atomically take the oldest runnable job. SKIP LOCKED lets parallel workers pass each other."""
+    now = utcnow()
+    job = db.scalar(
+        select(Job)
+        .where(Job.status == "queued", or_(Job.run_after.is_(None), Job.run_after <= now))
+        .order_by(Job.created_at)
+        .limit(1)
+        .with_for_update(skip_locked=True)
+    )
+    if not job:
+        db.rollback()
+        return None
+    job.status = "running"
+    job.attempts += 1
+    job.locked_by = worker_id
+    job.heartbeat_at = now
+    job.run_after = None
+    db.commit()
+    return job.id
+
+
+def backoff_seconds(attempt: int) -> int:
+    return min(30 * 2 ** max(attempt - 1, 0), 900)  # 30s, 60s, 120s ... capped at 15 min
+
+
+def retry_or_fail(db: Session, job: Job, error: str) -> bool:
+    """Requeue with backoff if attempts remain. Returns True if it will retry."""
+    if job.attempts < job.max_attempts:
+        wait = backoff_seconds(job.attempts)
+        job.status = "queued"
+        job.locked_by = None
+        job.run_after = utcnow() + timedelta(seconds=wait)
+        update_job(db, job, error=error[:1000], message=f"Retrying in {wait}s")
+        return True
+    update_job(db, job, status="failed", error=error[:1000], message="Failed")
+    return False
+
+
+def stale_jobs(db: Session) -> list[Job]:
+    cutoff = utcnow() - timedelta(seconds=settings.job_stale_seconds)
+    return list(
+        db.scalars(
+            select(Job)
+            .where(Job.status == "running", or_(Job.heartbeat_at.is_(None), Job.heartbeat_at < cutoff))
+            .with_for_update(skip_locked=True)
+        ).all()
+    )
+
+
+# Progress stream (API)
+
+
+async def watch_job(
+    session_factory: Callable[[], Session],
+    job_id: uuid.UUID,
+    interval: float = 1.0,
+    max_seconds: float = 4 * 3600,
+) -> AsyncIterator[str]:
+    """Yield the serialized job whenever it changes, until it finishes. Polls the row."""
+
+    def snapshot() -> dict | None:
+        with session_factory() as db:
+            job = db.get(Job, job_id)
+            return serialize_job(job) if job else None
+
+    last: dict | None = None
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + max_seconds
+    while loop.time() < deadline:
+        data = await run_in_threadpool(snapshot)
+        if data is None:
+            return
+        if data != last:
+            yield json.dumps(data)
+            last = data
+        if data["status"] in TERMINAL:
+            return
+        await asyncio.sleep(interval)
 
 
 def record_usage(

backend/app/worker.py

@@ -1,19 +1,32 @@
-"""arq worker: long running jobs + scheduled email batches.
+"""Worker: runs queued jobs from the Postgres jobs table plus scheduled email batches.
 
-Run: arq app.worker.WorkerSettings
+Run: python -m app.worker
+
+- Claims jobs with FOR UPDATE SKIP LOCKED, runs up to WORKER_CONCURRENCY at once in threads.
+- Failed jobs retry with backoff up to max_attempts; jobs whose worker died (no heartbeat for
+  JOB_STALE_SECONDS) are requeued.
+- Periodic tasks take a Postgres advisory lock, so with several workers only one runs each tick.
 """
 
 import logging
+import os
+import signal
+import socket
+import threading
+import time
 import uuid
+from collections.abc import Callable
+from concurrent.futures import Future, ThreadPoolExecutor
+from contextlib import contextmanager
+from dataclasses import dataclass, field
 from datetime import UTC, datetime
 
 import httpx
-from arq import cron
-from arq.connections import RedisSettings
-from sqlalchemy import select
+from sqlalchemy import select, text
+from sqlalchemy.orm import Session
 
 from app.config import settings
-from app.db import SessionLocal
+from app.db import SessionLocal, engine
 from app.models import (
     Artifact,
     Job,
@@ -28,65 +41,135 @@ from app.models import (
 from app.pipeline.ingest import ingest_source
 from app.services.email import email_service
 from app.services.email_verification import purge_old_codes
-from app.services.jobs import update_job
+from app.services.jobs import claim_next, retry_or_fail, stale_jobs, update_job
 from app.services.plans import effective_plan
 from app.services.storage import keys, storage
 
-log = logging.getLogger(__name__)
+log = logging.getLogger("notestack.worker")
+
+
+@dataclass
+class Done:
+    result: dict = field(default_factory=dict)
+    message: str = "Done"
+
+
+HANDED_OFF = object()  # the job continues elsewhere (renderer) and reports back via callbacks
+
+
+# Handlers: (db, job) -> Done | HANDED_OFF. Raise to trigger a retry.
+
+
+def handle_ingest(db: Session, job: Job):
+    source = db.get(Source, uuid.UUID(job.params["source_id"]))
+    if not source:
+        return Done({"error": "source deleted"}, "Source no longer exists")
+    plan = effective_plan(db, db.get(Workspace, source.workspace_id))
+    try:
+        result = ingest_source(db, source, job, max_posts=min(plan.indexed_posts, 200))
+    except Exception as exc:
+        db.rollback()
+        source.sync_status = "error"
+        source.sync_error = str(exc)[:500]
+        db.commit()
+        raise
+    return Done(result, "All posts in orbit")
 
 
-async def ingest_source_task(ctx, source_id: str, job_id: str) -> dict:
-    with SessionLocal() as db:
-        job = db.get(Job, uuid.UUID(job_id))
-        source = db.get(Source, uuid.UUID(source_id))
-        if not job or not source:
-            return {"error": "missing"}
+def handle_render(db: Session, job: Job):
+    """Hand off to the Remotion renderer with a presigned PUT so it never holds R2 credentials."""
+    p = job.params
+    artifact = db.get(Artifact, uuid.UUID(p["artifact_id"]))
+    ext = "png" if p["format"] == "png" else "mp4"
+    key = keys.artifact(artifact.workspace_id, artifact.id, p["composition"].lower(), ext)
+    content_type = "image/png" if ext == "png" else "video/mp4"
+    update_job(db, job, progress=0.01, message="T-minus: preparing render")
+    resp = httpx.post(
+        f"{settings.renderer_url}/render",
+        headers={"x-internal-token": settings.internal_token},
+        json={
+            "jobId": str(job.id),
+            "compositionId": p["composition"],
+            "props": p["props"],
+            "format": p["format"],
+            "uploadUrl": storage.presign_put(key, content_type, ttl=6 * 3600),
+            "uploadContentType": content_type,
+            "callbackUrl": f"{settings.api_url}/api/internal/jobs/{job.id}/progress",
+            "storageKey": key,
+        },
+        timeout=30,
+    )
+    resp.raise_for_status()
+    return HANDED_OFF
+
+
+HANDLERS: dict[str, Callable[[Session, Job], object]] = {
+    "ingest": handle_ingest,
+    "render": handle_render,
+}
+
+
+def run_job(
+    job_id: uuid.UUID,
+    session_factory: Callable[[], Session] = SessionLocal,
+    handlers: dict | None = None,
+) -> None:
+    handlers = handlers or HANDLERS
+    with session_factory() as db:
+        job = db.get(Job, job_id)
+        if not job:
+            return
+        handler = handlers.get(job.kind)
+        if not handler:
+            update_job(db, job, status="failed", error=f"No handler for job kind {job.kind!r}")
+            return
         try:
-            plan = effective_plan(db, db.get(Workspace, source.workspace_id))
-            result = ingest_source(db, source, job, max_posts=min(plan.indexed_posts, 200))
-            update_job(db, job, status="done", progress=1.0, message="All posts in orbit", result=result)
-            return result
+            outcome = handler(db, job)
         except Exception as exc:
-            log.exception("ingest failed source=%s", source_id)
+            log.exception("job %s (%s) attempt %s failed", job.id, job.kind, job.attempts)
             db.rollback()
-            source.sync_status = "error"
-            source.sync_error = str(exc)[:500]
-            update_job(db, job, status="failed", error=str(exc)[:1000], message="Ingestion failed")
-            return {"error": str(exc)}
+            job = db.get(Job, job_id)
+            retry_or_fail(db, job, f"{type(exc).__name__}: {exc}")
+            return
+        if outcome is HANDED_OFF:
+            return
+        done = outcome if isinstance(outcome, Done) else Done()
+        update_job(db, job, status="done", progress=1.0, message=done.message, result=done.result)
 
 
-async def render_task(ctx, artifact_id: str, job_id: str, composition: str, props: dict, fmt: str) -> dict:
-    """Hand off to the Remotion renderer with a presigned PUT so it never holds R2 credentials."""
-    with SessionLocal() as db:
-        job = db.get(Job, uuid.UUID(job_id))
-        artifact = db.get(Artifact, uuid.UUID(artifact_id))
-        ext = "png" if fmt == "png" else "mp4"
-        key = keys.artifact(artifact.workspace_id, artifact.id, composition.lower(), ext)
-        content_type = "image/png" if ext == "png" else "video/mp4"
-        update_job(db, job, status="running", progress=0.01, message="T-minus: preparing render")
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(
-                f"{settings.renderer_url}/render",
-                headers={"x-internal-token": settings.internal_token},
-                json={
-                    "jobId": job_id,
-                    "compositionId": composition,
-                    "props": props,
-                    "format": fmt,
-                    "uploadUrl": storage.presign_put(key, content_type, ttl=6 * 3600),
-                    "uploadContentType": content_type,
-                    "callbackUrl": f"{settings.api_url}/api/internal/jobs/{job_id}/progress",
-                    "storageKey": key,
-                },
-            )
-            resp.raise_for_status()
-        return {"accepted": True, "key": key}
+def recover_stale(session_factory: Callable[[], Session] = SessionLocal) -> int:
+    with session_factory() as db:
+        jobs = stale_jobs(db)
+        for job in jobs:
+            log.warning("job %s stale (worker %s), requeueing", job.id, job.locked_by)
+            retry_or_fail(db, job, "Worker stopped responding")
+        db.commit()
+        return len(jobs)
+
 
+# Periodic tasks
 
-async def update_email_batch(ctx) -> None:
-    """Hourly check; sends one daily batch per scheduled campaign at its send hour (UTC)."""
+
+@contextmanager
+def single_flight(name: str):
+    """Postgres advisory lock on a dedicated connection; yields False if another worker holds it."""
+    if engine.dialect.name != "postgresql":
+        yield True
+        return
+    with engine.connect() as conn:
+        got = conn.execute(text("SELECT pg_try_advisory_lock(hashtext(:n))"), {"n": name}).scalar()
+        try:
+            yield bool(got)
+        finally:
+            if got:
+                conn.execute(text("SELECT pg_advisory_unlock(hashtext(:n))"), {"n": name})
+                conn.commit()
+
+
+def update_email_batch(session_factory: Callable[[], Session] = SessionLocal) -> None:
+    """Sends one daily batch per scheduled campaign at its send hour (UTC)."""
     now = datetime.now(UTC)
-    with SessionLocal() as db:
+    with session_factory() as db:
         campaigns = db.scalars(select(UpdateEmail).where(UpdateEmail.status.in_(["scheduled", "running"]))).all()
         for campaign in campaigns:
             hour = campaign.send_hour if campaign.send_hour >= 0 else settings.update_email_send_hour
@@ -121,19 +204,74 @@ async def update_email_batch(ctx) -> None:
             db.commit()
 
 
-async def daily_cleanup(ctx) -> None:
-    with SessionLocal() as db:
+def daily_cleanup(session_factory: Callable[[], Session] = SessionLocal) -> None:
+    with session_factory() as db:
         purge_old_codes(db)
 
 
-async def startup(ctx) -> None:
-    logging.basicConfig(level=logging.INFO)
+@dataclass
+class Periodic:
+    name: str
+    every_seconds: float
+    fn: Callable[[], object]
+    next_at: float = 0.0
+
+
+PERIODIC = [
+    Periodic("recover_stale", 60, recover_stale),
+    Periodic("update_email_batch", 300, update_email_batch),
+    Periodic("daily_cleanup", 24 * 3600, daily_cleanup),
+]
+
+
+def run_periodic(now: float) -> None:
+    for task in PERIODIC:
+        if now < task.next_at:
+            continue
+        task.next_at = now + task.every_seconds
+        try:
+            with single_flight(task.name) as got:
+                if got:
+                    task.fn()
+        except Exception:
+            log.exception("periodic task %s failed", task.name)
+
+
+# Main loop
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
+    worker_id = f"{socket.gethostname()}:{os.getpid()}"
+    stop = threading.Event()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        signal.signal(sig, lambda *_: stop.set())
+
+    pool = ThreadPoolExecutor(max_workers=settings.worker_concurrency, thread_name_prefix="job")
+    running: set[Future] = set()
+    log.info("worker %s up, concurrency=%s", worker_id, settings.worker_concurrency)
+
+    while not stop.is_set():
+        running = {f for f in running if not f.done()}
+        run_periodic(time.monotonic())
+        claimed = False
+        while len(running) < settings.worker_concurrency and not stop.is_set():
+            try:
+                with SessionLocal() as db:
+                    job_id = claim_next(db, worker_id)
+            except Exception:
+                log.exception("claim failed")
+                break
+            if not job_id:
+                break
+            running.add(pool.submit(run_job, job_id))
+            claimed = True
+        if not claimed:
+            stop.wait(settings.worker_poll_seconds)
+
+    log.info("worker %s stopping, waiting for %d running jobs", worker_id, len(running))
+    pool.shutdown(wait=True)
 
 
-class WorkerSettings:
-    functions = [ingest_source_task, render_task]
-    cron_jobs = [cron(update_email_batch, minute=5), cron(daily_cleanup, hour=3, minute=17)]
-    redis_settings = RedisSettings.from_dsn(settings.redis_url)
-    on_startup = startup
-    max_jobs = 10
-    job_timeout = 60 * 30
+if __name__ == "__main__":
+    main()

backend/requirements.txt

@@ -13,8 +13,6 @@ requests==2.*
 boto3==1.35.*
 resend==2.*
 httpx==0.28.*
-arq==0.26.*
-redis==5.*
 dspy==2.6.*
 feedparser==6.*
 beautifulsoup4==4.*

backend/tests/conftest.py

@@ -31,7 +31,7 @@ def db_session():
 def client(db_session):
     app.dependency_overrides[get_db] = lambda: db_session
     ConsoleEmailProvider.sent.clear()
-    yield TestClient(app)  # no context manager: skips lifespan (no Redis needed)
+    yield TestClient(app)  # no context manager: skips lifespan
     app.dependency_overrides.clear()
 
 

backend/tests/test_queue.py

@@ -0,0 +1,139 @@
+import asyncio
+import json
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+import app.models  # noqa: F401
+from app.db import Base
+from app.models import Job, Workspace
+from app.services import jobs as q
+from app.worker import HANDED_OFF, Done, recover_stale, run_job
+
+
+@pytest.fixture()
+def factory():
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    return sessionmaker(bind=engine, expire_on_commit=False)
+
+
+@pytest.fixture()
+def ws(factory):
+    with factory() as db:
+        w = Workspace(name="w", owner_id=uuid.uuid4())
+        db.add(w)
+        db.commit()
+        return w.id
+
+
+def enqueue(factory, ws, kind="ingest", **kw):
+    with factory() as db:
+        return q.create_job(db, ws, kind, {"n": 1}, **kw).id
+
+
+def get(factory, job_id) -> Job:
+    with factory() as db:
+        return db.get(Job, job_id)
+
+
+def test_claims_oldest_first_and_only_once(factory, ws):
+    a, b = enqueue(factory, ws), enqueue(factory, ws)
+    with factory() as db:
+        assert q.claim_next(db, "w1") == a
+        assert q.claim_next(db, "w1") == b
+        assert q.claim_next(db, "w1") is None
+    job = get(factory, a)
+    assert job.status == "running" and job.attempts == 1 and job.locked_by == "w1"
+
+
+def test_success_marks_done(factory, ws):
+    jid = enqueue(factory, ws)
+    with factory() as db:
+        q.claim_next(db, "w1")
+    run_job(jid, factory, {"ingest": lambda db, job: Done({"indexed": 3}, "All posts in orbit")})
+    job = get(factory, jid)
+    assert (job.status, job.progress, job.result, job.message) == ("done", 1.0, {"indexed": 3}, "All posts in orbit")
+
+
+def test_failure_retries_with_backoff_then_fails(factory, ws):
+    jid = enqueue(factory, ws, max_attempts=2)
+
+    def boom(db, job):
+        raise RuntimeError("feed timed out")
+
+    with factory() as db:
+        q.claim_next(db, "w1")
+    run_job(jid, factory, {"ingest": boom})
+    job = get(factory, jid)
+    assert job.status == "queued" and "feed timed out" in job.error and job.message == "Retrying in 30s"
+    with factory() as db:
+        assert q.claim_next(db, "w1") is None  # backoff not elapsed yet
+        db.get(Job, jid).run_after = datetime.now(UTC) - timedelta(seconds=1)
+        db.commit()
+        assert q.claim_next(db, "w1") == jid
+    run_job(jid, factory, {"ingest": boom})
+    job = get(factory, jid)
+    assert job.status == "failed" and job.attempts == 2 and job.finished_at is not None
+
+
+def test_handed_off_job_stays_running(factory, ws):
+    jid = enqueue(factory, ws, kind="render")
+    with factory() as db:
+        q.claim_next(db, "w1")
+    run_job(jid, factory, {"render": lambda db, job: HANDED_OFF})
+    assert get(factory, jid).status == "running"
+
+
+def test_unknown_kind_fails(factory, ws):
+    jid = enqueue(factory, ws, kind="mystery")
+    with factory() as db:
+        q.claim_next(db, "w1")
+    run_job(jid, factory, {})
+    assert get(factory, jid).status == "failed"
+
+
+def test_stale_running_job_is_requeued(factory, ws, monkeypatch):
+    jid = enqueue(factory, ws)
+    with factory() as db:
+        q.claim_next(db, "dead-worker")
+        db.get(Job, jid).heartbeat_at = datetime.now(UTC) - timedelta(hours=1)
+        db.commit()
+    assert recover_stale(factory) == 1
+    job = get(factory, jid)
+    assert job.status == "queued" and job.error == "Worker stopped responding" and job.locked_by is None
+
+
+def test_progress_updates_count_as_heartbeat(factory, ws):
+    jid = enqueue(factory, ws)
+    with factory() as db:
+        q.claim_next(db, "w1")
+        job = db.get(Job, jid)
+        job.heartbeat_at = datetime.now(UTC) - timedelta(hours=1)
+        db.commit()
+        q.update_job(db, job, progress=0.5, message="halfway")
+    assert recover_stale(factory) == 0
+
+
+def test_watch_job_streams_changes_until_done(factory, ws):
+    jid = enqueue(factory, ws)
+
+    async def collect():
+        seen = []
+        async for data in q.watch_job(factory, jid, interval=0.01):
+            seen.append(json.loads(data))
+            with factory() as db:
+                job = db.get(Job, jid)
+                if len(seen) == 1:
+                    q.update_job(db, job, status="running", progress=0.5, message="halfway")
+                elif len(seen) == 2:
+                    q.update_job(db, job, status="done", progress=1.0)
+        return seen
+
+    seen = asyncio.run(collect())
+    assert [s["status"] for s in seen] == ["queued", "running", "done"]
+    assert seen[1]["message"] == "halfway"

backend/tests/test_queue_postgres.py

@@ -0,0 +1,71 @@
+"""Real Postgres checks for the parts SQLite cannot exercise: SKIP LOCKED under concurrency and
+advisory locks. Runs when TEST_POSTGRES_URL is set (CI provides one)."""
+
+import os
+import threading
+import uuid
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+import app.models  # noqa: F401
+from app.db import Base
+from app.models import Job, Workspace
+from app.services import jobs as q
+
+PG_URL = os.environ.get("TEST_POSTGRES_URL")
+pytestmark = pytest.mark.skipif(not PG_URL, reason="TEST_POSTGRES_URL not set")
+
+
+@pytest.fixture()
+def factory():
+    engine = create_engine(PG_URL, pool_size=20)
+    Base.metadata.drop_all(engine)
+    Base.metadata.create_all(engine)
+    yield sessionmaker(bind=engine, expire_on_commit=False)
+    Base.metadata.drop_all(engine)
+    engine.dispose()
+
+
+def test_parallel_workers_never_claim_the_same_job(factory):
+    with factory() as db:
+        ws = Workspace(name="w", owner_id=uuid.uuid4())
+        db.add(ws)
+        db.commit()
+        ids = {q.create_job(db, ws.id, "ingest").id for _ in range(60)}
+
+    claimed: list[uuid.UUID] = []
+    lock = threading.Lock()
+
+    def worker(name):
+        while True:
+            with factory() as db:
+                jid = q.claim_next(db, name)
+            if not jid:
+                return
+            with lock:
+                claimed.append(jid)
+
+    threads = [threading.Thread(target=worker, args=(f"w{i}",)) for i in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    assert len(claimed) == len(set(claimed)) == len(ids)
+    with factory() as db:
+        assert db.query(Job).filter(Job.attempts != 1).count() == 0
+
+
+def test_single_flight_excludes_second_holder(factory, monkeypatch):
+    import app.worker as worker
+
+    engine = factory.kw["bind"]
+    monkeypatch.setattr(worker, "engine", engine)
+    with worker.single_flight("email-batch") as first:
+        with worker.single_flight("email-batch") as second:
+            assert first is True and second is False
+    with worker.single_flight("email-batch") as again:
+        assert again is True
+    with engine.connect() as c:
+        assert c.execute(text("select count(*) from pg_locks where locktype = 'advisory'")).scalar() == 0

docker-compose.yml

@@ -13,10 +13,6 @@ services:
       interval: 5s
       retries: 10
 
-  redis:
-    image: redis:7-alpine
-    ports: ["6379:6379"]
-
   minio:
     image: minio/minio:latest
     command: server /data --console-address ":9001"
@@ -31,7 +27,6 @@ services:
     env_file: .env
     environment:
       DATABASE_URL: postgresql+psycopg://notestack:notestack@postgres:5432/notestack
-      REDIS_URL: redis://redis:6379/0
       R2_ENDPOINT_URL: ${R2_ENDPOINT_URL:-http://minio:9000}
       RENDERER_URL: http://renderer:3100
       API_URL: http://api:8000
@@ -40,15 +35,13 @@ services:
     ports: ["8000:8000"]
     depends_on:
       postgres: { condition: service_healthy }
-      redis: { condition: service_started }
 
   worker:
     build: ./backend
-    command: arq app.worker.WorkerSettings
+    command: python -m app.worker
     env_file: .env
     environment:
       DATABASE_URL: postgresql+psycopg://notestack:notestack@postgres:5432/notestack
-      REDIS_URL: redis://redis:6379/0
       R2_ENDPOINT_URL: ${R2_ENDPOINT_URL:-http://minio:9000}
       RENDERER_URL: http://renderer:3100
       API_URL: http://api:8000

docs/DESIGN.md

@@ -26,7 +26,7 @@ Every factual statement Notestack generates must trace back to the writer's own
 | Area | Decision |
 | --- | --- |
 | Backend | FastAPI + SQLAlchemy 2 + Alembic (same stack as blog2video) |
-| Worker | arq on Redis (async jobs, progress events over Redis pub/sub, SSE to clients) |
+| Worker | Postgres backed queue: the `jobs` table, claimed with `FOR UPDATE SKIP LOCKED`; progress on the row, streamed as SSE. No Redis. |
 | Database | Postgres 16 (no vector extension) |
 | Frontend | React + Vite + TypeScript, plain CSS with design tokens |
 | Renderer | Node service running Remotion (`renderer/`) |
@@ -45,9 +45,9 @@ Every factual statement Notestack generates must trace back to the writer's own
             +-------------------+                          |  auth, notebooks,  |
                      ^                                     |  billing, storage  |
                      | presigned GET                       +---------+----------+
-                     |                                               | enqueue (arq)
+                     |                                               | insert queued job
             +--------+----------+                          +---------v----------+
-            | Cloudflare R2     |<-------------------------|  worker (arq)      |
+            | Cloudflare R2     |<-------------------------|  worker            |
             | notestack-assets  |   put objects            |  ingest, corpus,   |
             +--------+----------+                          |  DSPy, TTS         |
                      ^                                     +---------+----------+
@@ -55,7 +55,7 @@ Every factual statement Notestack generates must trace back to the writer's own
             +--------+----------+   progress callbacks     +---------v----------+
             | renderer (Remotion)|------------------------>|  api /internal/... |
             +-------------------+                          +--------------------+
-     Postgres (api, worker)        Redis (queue, pub/sub, cache)
+     Postgres: data + job queue (api, worker)
 ```
 
 ### 3.1 Storage pipeline (R2)
@@ -159,11 +159,28 @@ Z.ai considerations baked in:
 
 Swapping to Anthropic, OpenAI, Gemini or a local model is only an env change.
 
-### 3.4 Jobs and progress
-
-Every long operation is a row in `jobs` (kind, status, progress, error, cost) plus an arq task.
-Workers publish progress to Redis channel `job:{id}`; `GET /api/jobs/{id}/events` streams it as SSE.
-No API request blocks for more than a few seconds.
+### 3.4 Jobs and progress (Postgres, no Redis)
+
+The `jobs` table is the queue, the progress record and the SSE source.
+
+- **Enqueue:** the API inserts a `queued` row (`create_job`). Nothing else.
+- **Claim:** `python -m app.worker` polls every `WORKER_POLL_SECONDS` and takes the oldest runnable
+  row with `SELECT ... FOR UPDATE SKIP LOCKED`, so any number of workers can share the table without
+  double running a job. Up to `WORKER_CONCURRENCY` jobs run at once in threads.
+- **Retry:** a handler that raises is requeued with backoff (30s, 60s, 120s... capped at 15 min)
+  until `max_attempts` (default 3), then marked `failed`.
+- **Heartbeat:** every progress update stamps `heartbeat_at`. A `running` job silent for
+  `JOB_STALE_SECONDS` (worker crashed, renderer died) is requeued by the stale sweeper.
+- **Hand off:** render jobs return `HANDED_OFF`; the renderer's progress callbacks keep the row
+  alive and mark it done.
+- **Progress to the browser:** `GET /api/jobs/{id}/events` re-reads the row about once a second and
+  sends only changes. Latency is at most a second, which is fine for progress bars.
+- **Schedules:** the worker runs the stale sweep (1 min), update email batch (5 min, gated to the
+  campaign send hour) and daily cleanup. Each takes a Postgres advisory lock, so with several
+  workers only one runs a given task.
+
+No request blocks for more than a few seconds. Add Redis back only if job volume makes polling a
+measurable load on Postgres; the handler code would not change.
 
 ## 4. Data model
 

docs/PHASE0.md

@@ -4,9 +4,9 @@ Legend: [x] done in this scaffold, [ ] open
 
 ## Repo and infra
 - [x] Monorepo layout: `backend/`, `frontend/`, `renderer/`, `docs/`
-- [x] docker-compose: api, worker, postgres, redis, minio (local R2 stand-in), renderer, frontend
+- [x] docker-compose: api, worker, postgres, minio (local R2 stand-in), renderer, frontend
 - [x] `.env.example` covering every service
-- [x] CI: backend lint + tests, frontend typecheck + build, renderer typecheck
+- [x] CI: backend lint + tests (with a Postgres service), frontend typecheck + build, renderer typecheck
 - [ ] Production deploy target (Fly, Render or DO App Platform) and R2 bucket + custom domain
 - [ ] Sentry (api, worker, frontend)
 
@@ -14,7 +14,8 @@ Legend: [x] done in this scaffold, [ ] open
 - [x] Settings (`app/config.py`), DB session, Alembic initial migration
 - [x] Full data model from design doc Section 4
 - [x] Health endpoint, CORS, request id logging
-- [x] Jobs table + arq worker + Redis progress pub/sub + SSE endpoint
+- [x] Postgres job queue (SKIP LOCKED claims, retries with backoff, stale recovery) + SSE progress
+- [x] Scheduled tasks in the worker behind advisory locks
 - [x] Usage metering table and helper (`record_usage`)
 
 ## Auth (mirrors blog2video)