diff --git a/backend/migrations/versions/8d51af959ae4_initial_schema.py b/backend/migrations/versions/8d51af959ae4_initial_schema.py index d12331d..ace87d1 100644 --- a/backend/migrations/versions/8d51af959ae4_initial_schema.py +++ b/backend/migrations/versions/8d51af959ae4_initial_schema.py @@ -19,6 +19,13 @@ branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None +def _json_type() -> sa.types.TypeEngine: + """JSONB on PostgreSQL, plain JSON on SQLite (desktop bundled engine).""" + if op.get_context().dialect.name == "postgresql": + return postgresql.JSONB(astext_type=sa.Text()) + return sa.JSON() + + def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### op.create_table('users', @@ -81,7 +88,7 @@ def upgrade() -> None: ) op.create_index(op.f('ix_recordings_client_recording_id'), 'recordings', ['client_recording_id'], unique=False) op.create_index(op.f('ix_recordings_processing_status'), 'recordings', ['processing_status'], unique=False) - op.create_index('ix_recordings_user_client_id', 'recordings', ['user_id', 'client_recording_id'], unique=True, postgresql_where='client_recording_id IS NOT NULL') + op.create_index('ix_recordings_user_client_id', 'recordings', ['user_id', 'client_recording_id'], unique=True, postgresql_where='client_recording_id IS NOT NULL', sqlite_where=sa.text('client_recording_id IS NOT NULL')) op.create_index(op.f('ix_recordings_user_id'), 'recordings', ['user_id'], unique=False) op.create_index('ix_recordings_user_recorded', 'recordings', ['user_id', 'recorded_at'], unique=False) op.create_table('refresh_tokens', @@ -120,7 +127,7 @@ def upgrade() -> None: ) op.create_index(op.f('ix_assets_recording_id'), 'assets', ['recording_id'], unique=False) op.create_index(op.f('ix_assets_user_id'), 'assets', ['user_id'], unique=False) - op.create_index('uq_assets_one_original', 'assets', ['recording_id'], unique=True, postgresql_where="kind = 'original'") + op.create_index('uq_assets_one_original', 'assets', ['recording_id'], unique=True, postgresql_where="kind = 'original'", sqlite_where=sa.text("kind = 'original'")) op.create_table('processing_jobs', sa.Column('recording_id', sa.Uuid(), nullable=False), sa.Column('job_type', sa.Enum('normalize_audio', 'transcribe', 'summarize', name='job_type'), nullable=False), @@ -153,7 +160,7 @@ def upgrade() -> None: sa.Column('superseded_at', sa.DateTime(timezone=True), nullable=True), sa.Column('provider', sa.String(length=80), nullable=False), sa.Column('model', sa.String(length=120), nullable=True), - sa.Column('content', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('content', _json_type(), nullable=False), sa.Column('edited_by_user', sa.Boolean(), nullable=False), sa.Column('id', sa.Uuid(), nullable=False), sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), @@ -170,7 +177,7 @@ def upgrade() -> None: sa.Column('provider', sa.String(length=80), nullable=False), sa.Column('model', sa.String(length=120), nullable=True), sa.Column('text', sa.Text(), nullable=False), - sa.Column('segments', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('segments', _json_type(), nullable=True), sa.Column('edited_by_user', sa.Boolean(), nullable=False), sa.Column('id', sa.Uuid(), nullable=False), sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), diff --git a/backend/migrations/versions/fts0000000001_fts_columns.py b/backend/migrations/versions/fts0000000001_fts_columns.py index 576b003..4257d64 100644 --- a/backend/migrations/versions/fts0000000001_fts_columns.py +++ b/backend/migrations/versions/fts0000000001_fts_columns.py @@ -20,6 +20,11 @@ depends_on: str | Sequence[str] | None = None def upgrade() -> None: + # tsvector is PostgreSQL-only. On SQLite (desktop bundled engine) full- + # text search is handled outside the DB (the app scans report files), + # so this migration is a no-op. + if op.get_context().dialect.name != "postgresql": + return # recordings: title + notes + tag names are indexed; transcript/summary # contribute via their own tables (joined at query time). op.execute( @@ -74,6 +79,8 @@ def upgrade() -> None: def downgrade() -> None: + if op.get_context().dialect.name != "postgresql": + return for table in ("tags", "summaries", "transcripts", "recordings"): op.execute(f"DROP INDEX IF EXISTS ix_{table}_search;") op.execute(f"ALTER TABLE {table} DROP COLUMN IF EXISTS search_vector;") diff --git a/backend/migrations/versions/m8models000001_model_support.py b/backend/migrations/versions/m8models000001_model_support.py index c984e3c..37aa5ed 100644 --- a/backend/migrations/versions/m8models000001_model_support.py +++ b/backend/migrations/versions/m8models000001_model_support.py @@ -15,12 +15,19 @@ import sqlalchemy as sa from alembic import op from sqlalchemy.dialects import postgresql -revision: str = "m8models000001" -down_revision: str | None = "fts0000000001" +revision: str = 'm8models000001' +down_revision: str | None = 'fts0000000001' branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None +def _json_type() -> sa.types.TypeEngine: + """JSONB on PostgreSQL, plain JSON on SQLite (desktop bundled engine).""" + if op.get_context().dialect.name == "postgresql": + return postgresql.JSONB(astext_type=sa.Text()) + return sa.JSON() + + def upgrade() -> None: op.add_column( "recordings", @@ -43,9 +50,8 @@ def upgrade() -> None: sa.Column("key", sa.String(length=120), nullable=False), sa.Column( "value", - postgresql.JSONB(astext_type=sa.Text()), + _json_type(), nullable=False, - server_default=sa.text("'{}'::jsonb"), ), sa.Column("id", sa.Uuid(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), diff --git a/backend/shonar/core/config.py b/backend/shonar/core/config.py index eb50cf0..f67a03a 100644 --- a/backend/shonar/core/config.py +++ b/backend/shonar/core/config.py @@ -64,6 +64,14 @@ class Settings(BaseSettings): # --- Worker / queue ------------------------------------------------------- redis_url: str = "redis://localhost:6379/0" + # "arq" = Redis transport (server deployments). "inline" = in-process + # asyncio runner (desktop bundled-lite engine: no Redis, one job at a + # time). DB rows are the queue of record either way. + queue_backend: str = "arq" # arq | inline + # Apply 'alembic upgrade head' at startup. The desktop bundled engine + # owns its SQLite file and must self-migrate; server deployments run + # migrations in their deploy flow instead. + auto_migrate: bool = False # --- Audio processing ----------------------------------------------------- # Optional server-side conversion via FFmpeg. Off by default. diff --git a/backend/shonar/db/base.py b/backend/shonar/db/base.py index d8d01e8..ab9980d 100644 --- a/backend/shonar/db/base.py +++ b/backend/shonar/db/base.py @@ -5,10 +5,27 @@ from __future__ import annotations import uuid from datetime import UTC, datetime -from sqlalchemy import DateTime, Uuid +from sqlalchemy import DateTime, TypeDecorator, Uuid from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +class UTCDT(TypeDecorator): + """Timezone-aware datetime that survives SQLite round-trips. + + Postgres returns aware values (impl is a no-op there); SQLite has no + tz info and hands back naive datetimes, which crash comparisons with + ``utcnow()``. Attach UTC on read whenever the driver lost it. + """ + + impl = DateTime(timezone=True) + cache_ok = True + + def process_result_value(self, value, dialect): # noqa: ANN001, ANN201 + if value is not None and value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value + + def utcnow() -> datetime: return datetime.now(UTC) @@ -25,9 +42,9 @@ class PublicIdMixin: """Gives every table a UUID public id; integer PKs stay internal.""" id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=new_uuid) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), default=utcnow, nullable=False - ) + created_at: Mapped[datetime] = mapped_column(UTCDT(), default=utcnow, nullable=False) + # default= as well as onupdate=: SQLite enforces NOT NULL on insert + # (Postgres silently stored NULL for rows that never got an UPDATE). updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False + UTCDT(), default=utcnow, onupdate=utcnow, nullable=False ) diff --git a/backend/shonar/db/migrate.py b/backend/shonar/db/migrate.py new file mode 100644 index 0000000..f297f16 --- /dev/null +++ b/backend/shonar/db/migrate.py @@ -0,0 +1,33 @@ +"""Programmatic schema migrations for the bundled desktop engine. + +Server deployments run ``alembic upgrade head`` in their deploy flow; +the desktop app owns its SQLite file end-to-end, so the API applies +migrations itself at startup (``SHONAR_AUTO_MIGRATE=1``). +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path + +from alembic import command +from alembic.config import Config + +logger = logging.getLogger("shonar.migrate") + +# backend/ dir: migrations/ + alembic.ini live here relative to this file. +_BACKEND_DIR = Path(__file__).resolve().parents[2] + + +async def upgrade_head(database_url: str) -> None: + """Run 'alembic upgrade head' for `database_url` without blocking the loop.""" + + def _run() -> None: + cfg = Config(str(_BACKEND_DIR / "alembic.ini")) + cfg.set_main_option("script_location", str(_BACKEND_DIR / "migrations")) + cfg.set_main_option("sqlalchemy.url", database_url) + command.upgrade(cfg, "head") + + await asyncio.to_thread(_run) + logger.info("database schema migrated to head") diff --git a/backend/shonar/db/models.py b/backend/shonar/db/models.py index 2a78dea..a19f4c2 100644 --- a/backend/shonar/db/models.py +++ b/backend/shonar/db/models.py @@ -13,10 +13,12 @@ import enum import uuid from datetime import datetime +# JSON that renders JSONB on PostgreSQL and plain JSON on SQLite (the +# desktop bundled-lite engine runs on SQLite; JSONB does not compile there). +from sqlalchemy import JSON as _JSON from sqlalchemy import ( BigInteger, Boolean, - DateTime, Enum, Float, ForeignKey, @@ -25,17 +27,12 @@ from sqlalchemy import ( Text, UniqueConstraint, ) +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship -from shonar.db.base import Base, PublicIdMixin, utcnow +from shonar.db.base import UTCDT, Base, PublicIdMixin, utcnow -# JSON type that uses JSONB on PostgreSQL and plain JSON elsewhere (tests on -# SQLite would need JSONB emulation; we run tests against Postgres, so JSONB -# is fine, with a generic JSON fallback for ad-hoc SQLite use). -try: # pragma: no cover - from sqlalchemy.dialects.postgresql import JSONB as JSONType -except ImportError: # pragma: no cover - from sqlalchemy import JSON as JSONType # type: ignore[attr-defined,no-redef] +JSONType = _JSON().with_variant(JSONB(), "postgresql") # --------------------------------------------------------------------------- @@ -53,7 +50,7 @@ class User(Base, PublicIdMixin): # Feature switches the user controls (mirror of app settings, server truth # for e.g. whether location metadata may be stored at all). location_storage_enabled: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) - deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + deleted_at: Mapped[datetime | None] = mapped_column(UTCDT()) devices: Mapped[list[Device]] = relationship(back_populates="user") recordings: Mapped[list[Recording]] = relationship(back_populates="user") @@ -67,8 +64,8 @@ class Device(Base, PublicIdMixin): ) name: Mapped[str] = mapped_column(String(120), nullable=False) platform: Mapped[str] = mapped_column(String(40), default="android", nullable=False) - last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) - revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + last_seen_at: Mapped[datetime] = mapped_column(UTCDT(), default=utcnow) + revoked_at: Mapped[datetime | None] = mapped_column(UTCDT()) user: Mapped[User] = relationship(back_populates="devices") @@ -88,8 +85,8 @@ class RefreshToken(Base, PublicIdMixin): device_id: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("devices.id", ondelete="SET NULL") ) - expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) - revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + expires_at: Mapped[datetime] = mapped_column(UTCDT(), nullable=False) + revoked_at: Mapped[datetime | None] = mapped_column(UTCDT()) replaced_by: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("refresh_tokens.id", ondelete="SET NULL") ) @@ -124,7 +121,7 @@ class Recording(Base, PublicIdMixin): client_recording_id: Mapped[str | None] = mapped_column(String(64), index=True) title: Mapped[str] = mapped_column(String(300), nullable=False, default="Untitled recording") - recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + recorded_at: Mapped[datetime] = mapped_column(UTCDT(), nullable=False) duration_seconds: Mapped[float] = mapped_column(Float, default=0.0, nullable=False) notes: Mapped[str | None] = mapped_column(Text) # Location is stored ONLY when the user has explicitly enabled it. @@ -151,7 +148,7 @@ class Recording(Base, PublicIdMixin): # The original audio is the Asset row with kind=original for this # recording (at most one, enforced by a partial unique index). Keeping # the pointer one-directional avoids a recordings<->assets FK cycle. - deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + deleted_at: Mapped[datetime | None] = mapped_column(UTCDT()) user: Mapped[User] = relationship(back_populates="recordings") assets: Mapped[list[Asset]] = relationship( @@ -255,7 +252,7 @@ class UploadSession(Base, PublicIdMixin): default=UploadSessionStatus.open, nullable=False, ) - expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + expires_at: Mapped[datetime] = mapped_column(UTCDT(), nullable=False) completed_asset_id: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("assets.id", ondelete="SET NULL") ) @@ -276,7 +273,7 @@ class UploadChunk(Base): size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False) checksum_sha256: Mapped[str] = mapped_column(String(64), nullable=False) storage_key: Mapped[str] = mapped_column(String(500), nullable=False) - created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + created_at: Mapped[datetime] = mapped_column(UTCDT(), default=utcnow) session: Mapped[UploadSession] = relationship(back_populates="chunks") @@ -297,7 +294,7 @@ class Transcript(Base, PublicIdMixin): # Versioned: regenerated transcripts supersede older rows; the newest # non-superseded row is authoritative. ``edited_by_user`` rows win. version: Mapped[int] = mapped_column(default=1, nullable=False) - superseded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + superseded_at: Mapped[datetime | None] = mapped_column(UTCDT()) language: Mapped[str | None] = mapped_column(String(16)) provider: Mapped[str] = mapped_column(String(80), nullable=False, default="manual") model: Mapped[str | None] = mapped_column(String(120)) @@ -332,7 +329,7 @@ class Summary(Base, PublicIdMixin): ForeignKey("recordings.id", ondelete="CASCADE"), index=True, nullable=False ) version: Mapped[int] = mapped_column(default=1, nullable=False) - superseded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + superseded_at: Mapped[datetime | None] = mapped_column(UTCDT()) provider: Mapped[str] = mapped_column(String(80), nullable=False, default="manual") model: Mapped[str | None] = mapped_column(String(120)) content: Mapped[dict] = mapped_column(JSONType, nullable=False, default=dict) @@ -414,8 +411,8 @@ class ProcessingJob(Base, PublicIdMixin): stage: Mapped[str | None] = mapped_column(String(32)) # 0-100 work estimate within the current stage, when known. progress: Mapped[int | None] = mapped_column() - started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) - finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + started_at: Mapped[datetime | None] = mapped_column(UTCDT()) + finished_at: Mapped[datetime | None] = mapped_column(UTCDT()) # Opaque arq task handle for observability. task_handle: Mapped[str | None] = mapped_column(String(120)) diff --git a/backend/shonar/db/session.py b/backend/shonar/db/session.py index 595945a..bba2b68 100644 --- a/backend/shonar/db/session.py +++ b/backend/shonar/db/session.py @@ -18,16 +18,38 @@ def get_engine(): settings = get_settings() kwargs: dict = {"pool_pre_ping": True} url = settings.database_url - # SQLite (tests) does not support the pg pool sizing kwargs. + # SQLite (tests, desktop bundled engine) does not support the pg + # pool sizing kwargs. if url.startswith("sqlite"): kwargs = {} else: kwargs.update(pool_size=settings.db_pool_size, max_overflow=settings.db_max_overflow) _engine = create_async_engine(url, **kwargs) + if url.startswith("sqlite"): + _configure_sqlite(_engine) _session_factory = async_sessionmaker(_engine, expire_on_commit=False) return _engine +def _configure_sqlite(engine) -> None: # noqa: ANN001 + """Per-connection SQLite pragmas. + + foreign_keys is OFF by default in SQLite; the schema leans on + ON DELETE CASCADE, so every connection must enable it. WAL lets the + inline-queue writer and API readers coexist without SQLITE_BUSY + storms on the desktop box. + """ + from sqlalchemy import event + + @event.listens_for(engine.sync_engine, "connect") + def _pragmas(dbapi_conn, _record): # noqa: ANN001 + cur = dbapi_conn.cursor() + cur.execute("PRAGMA foreign_keys=ON") + cur.execute("PRAGMA journal_mode=WAL") + cur.execute("PRAGMA busy_timeout=5000") + cur.close() + + async def dispose_engine() -> None: global _engine, _session_factory if _engine is not None: diff --git a/backend/shonar/main.py b/backend/shonar/main.py index cbf7575..16fb2a8 100644 --- a/backend/shonar/main.py +++ b/backend/shonar/main.py @@ -29,7 +29,20 @@ async def lifespan(app: FastAPI): get_engine() # validate URL parses; connections are lazy for warning in settings.validate_production(): logger.warning("CONFIG: %s", warning) + if settings.auto_migrate: + from shonar.db.migrate import upgrade_head + + await upgrade_head(settings.database_url) + inline = settings.queue_backend.strip().lower() == "inline" + if inline: + from shonar.services import inline_queue + + await inline_queue.start() yield + if inline: + from shonar.services import inline_queue + + await inline_queue.stop() await dispose_engine() diff --git a/backend/shonar/services/inline_queue.py b/backend/shonar/services/inline_queue.py new file mode 100644 index 0000000..a09e6cf --- /dev/null +++ b/backend/shonar/services/inline_queue.py @@ -0,0 +1,92 @@ +"""In-process job runner (desktop bundled-lite engine). + +``queue_backend=inline`` replaces the arq/Redis transport with a single +asyncio consumer inside the uvicorn process: DB rows stay the source of +truth (ProcessingJob), this just runs the work. One job at a time — +local faster-whisper is multi-GB per pass, mirroring the worker's +``max_jobs=1`` rule. + +Transient failures retry in-process with backoff up to MAX_TRIES (the +same budget arq gives via ``max_tries``); the final failure is recorded +by the task body itself (see processing._fail). +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging + +from shonar.db.models import JobType +from shonar.services import processing + +logger = logging.getLogger("shonar.inline_queue") + +RETRY_DELAY_SECONDS = 5.0 + +_queue: asyncio.Queue[tuple[str, str, int]] | None = None +_consumer: asyncio.Task | None = None + + +def _get_queue() -> asyncio.Queue[tuple[str, str, int]]: + global _queue + if _queue is None: + _queue = asyncio.Queue() + return _queue + + +async def start() -> None: + """Start the consumer and re-run anything the DB says is pending.""" + global _consumer + _get_queue() + if _consumer is None or _consumer.done(): + _consumer = asyncio.create_task(_consume(), name="shonar-inline-queue") + # Crash recovery: queued rows (and orphaned running rows requeued by + # the sweep) go back on the in-process queue. + count = await processing.sweep_stale() + if count: + logger.info("inline queue startup sweep requeued %d jobs", count) + + +async def stop() -> None: + global _consumer + if _consumer is not None: + _consumer.cancel() + with contextlib.suppress(BaseException): # noqa: BLE001 — shutdown is best-effort + await _consumer + _consumer = None + + +async def enqueue(job_type: JobType, recording_id: str) -> None: + await _get_queue().put((job_type.value, str(recording_id), 1)) + + +async def _consume() -> None: + q = _get_queue() + while True: + job_value, recording_id, attempt = await q.get() + ctx = {"job_try": attempt} + try: + if job_value == JobType.transcribe.value: + await processing.run_transcribe(ctx, recording_id) + else: + await processing.run_summarize(ctx, recording_id) + except processing.ProviderTransientError as e: + if attempt < processing.MAX_TRIES: + logger.warning( + "inline job %s:%s transient failure (%s); retry %d/%d", + job_value, recording_id, e, attempt + 1, processing.MAX_TRIES, + ) + await asyncio.sleep(RETRY_DELAY_SECONDS) + await q.put((job_value, recording_id, attempt + 1)) + else: + logger.error( + "inline job %s:%s failed after %d tries: %s", + job_value, recording_id, attempt, e, + ) + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 — consumer must survive any task crash + logger.exception("inline job %s:%s crashed", job_value, recording_id) + finally: + q.task_done() diff --git a/backend/shonar/services/processing.py b/backend/shonar/services/processing.py index e31061f..8090cac 100644 --- a/backend/shonar/services/processing.py +++ b/backend/shonar/services/processing.py @@ -144,8 +144,14 @@ async def latest_transcript_text( async def transport_enqueue(job_type: JobType, recording_id: uuid.UUID) -> None: - """Best-effort arq trigger. Failure only logs: the DB rows are the real - queue and the worker sweep picks up anything the transport missed.""" + """Best-effort trigger for the configured backend ('arq' or 'inline'). + Failure only logs: the DB rows are the real queue and the sweep picks + up anything the transport missed.""" + if get_settings().queue_backend.strip().lower() == "inline": + from shonar.services import inline_queue + + await inline_queue.enqueue(job_type, str(recording_id)) + return from arq import create_pool from arq.connections import RedisSettings from arq.constants import result_key_prefix diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 96c9053..dbeb693 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -48,20 +48,26 @@ async def _setup_db() -> AsyncIterator[None]: async def clean_db(_setup_db: None) -> AsyncIterator[None]: """Truncate between tests for isolation.""" yield - from sqlalchemy import text + from sqlalchemy import delete, text - from shonar.db.session import _session_factory # type: ignore[attr-defined] + from shonar.db.base import Base + from shonar.db.session import _session_factory, get_engine # type: ignore[attr-defined] assert _session_factory is not None async with _session_factory() as s: - await s.execute( - text( - "TRUNCATE users, devices, refresh_tokens, recordings, assets, " - "upload_sessions, upload_chunks, transcripts, summaries, tags, " - "recording_tags, processing_jobs, export_jobs, app_settings " - "RESTART IDENTITY CASCADE" + if get_engine().dialect.name == "postgresql": + await s.execute( + text( + "TRUNCATE users, devices, refresh_tokens, recordings, assets, " + "upload_sessions, upload_chunks, transcripts, summaries, tags, " + "recording_tags, processing_jobs, export_jobs, app_settings " + "RESTART IDENTITY CASCADE" + ) ) - ) + else: + # SQLite: delete child-first (reverse dependency order). + for table in reversed(Base.metadata.sorted_tables): + await s.execute(delete(table)) await s.commit() diff --git a/backend/tests/test_inline_queue.py b/backend/tests/test_inline_queue.py new file mode 100644 index 0000000..a0888f6 --- /dev/null +++ b/backend/tests/test_inline_queue.py @@ -0,0 +1,101 @@ +"""Inline queue tests (desktop bundled-lite engine): with +``queue_backend=inline`` an upload runs transcribe→summarize automatically +in-process — no arq, no Redis. Providers are fakes (see test_ai_pipeline). +""" + +from __future__ import annotations + +import asyncio +import uuid + +from sqlalchemy import select + +from shonar.core.config import get_settings +from shonar.db import session as db_session +from shonar.db.models import JobStatus, JobType, ProcessingJob +from shonar.services import inline_queue, processing + +from .test_ai_pipeline import FakeLlm, FakeTranscriber, upload_recording, use_fakes, user_tokens + + +def use_inline(monkeypatch): + """Force queue_backend=inline for everything resolved via processing.""" + inline_settings = get_settings().model_copy(update={"queue_backend": "inline"}) + monkeypatch.setattr(processing, "get_settings", lambda: inline_settings) + + +async def _wait_terminal(recording_id: str, want: int = 2, timeout: float = 20.0): + """Poll the DB until `want` jobs reach a terminal state.""" + deadline = asyncio.get_running_loop().time() + timeout + rows: list[ProcessingJob] = [] + while asyncio.get_running_loop().time() < deadline: + async with db_session._session_factory() as s: + rows = list( + ( + await s.scalars( + select(ProcessingJob).where( + ProcessingJob.recording_id == uuid.UUID(recording_id) + ) + ) + ).all() + ) + if sum( + 1 + for j in rows + if j.status in (JobStatus.succeeded, JobStatus.failed, JobStatus.skipped) + ) >= want: + return {j.job_type: j.status for j in rows} + await asyncio.sleep(0.1) + raise AssertionError( + f"jobs did not reach terminal state in {timeout}s: " + f"{[(j.job_type, j.status) for j in rows]}" + ) + + +async def test_inline_backend_processes_upload_without_arq(client, monkeypatch): + use_fakes(monkeypatch, FakeTranscriber(), FakeLlm()) + use_inline(monkeypatch) + await inline_queue.start() + try: + token = await user_tokens(client, email="inline@shonar.dev") + rec = await upload_recording(client, token, client_id="inline-1") + + statuses = await _wait_terminal(rec["id"], want=2) + assert statuses[JobType.transcribe] == JobStatus.succeeded + assert statuses[JobType.summarize] == JobStatus.succeeded + + h = {"Authorization": f"Bearer {token}"} + r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h) + assert r.json()["processing_status"] == "completed" + t = await client.get(f"/api/v1/recordings/{rec['id']}/transcript", headers=h) + assert t.json()["text"] == "hello world from the meeting" + s = await client.get(f"/api/v1/recordings/{rec['id']}/summary", headers=h) + assert s.json()["content"]["short"] == "Standup happened." + finally: + await inline_queue.stop() + + +async def test_inline_retries_transient_failure(client, monkeypatch): + stt = FakeTranscriber() + state = {"tries": 0} + original = stt.transcribe + + async def flaky(audio, mime, *, language_hint=None): + state["tries"] += 1 + if state["tries"] == 1: + raise processing.ProviderTransientError("temporary hiccup") + return await original(audio, mime, language_hint=language_hint) + + stt.transcribe = flaky + use_fakes(monkeypatch, stt, FakeLlm()) + use_inline(monkeypatch) + monkeypatch.setattr(inline_queue, "RETRY_DELAY_SECONDS", 0.05) + await inline_queue.start() + try: + token = await user_tokens(client, email="inline2@shonar.dev") + rec = await upload_recording(client, token, client_id="inline-2") + statuses = await _wait_terminal(rec["id"], want=2) + assert statuses[JobType.transcribe] == JobStatus.succeeded + assert state["tries"] >= 2 # the retry actually happened + finally: + await inline_queue.stop()