From 3aa1ad77313939203e000ccf5b14a5943ddb14f2 Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 22:49:10 +0300 Subject: [PATCH 1/7] feat(infra): add pgbouncer, minio lifecycle, updated_at and dash Add pgbouncer service in transaction mode with pool tuning, minio-setup ilm rule for raw/ 7d expiry via mc ilm rule add, videos.updated_at column with trigger and migration 000003, nginx-vod dash location and hls mapping fix, env KEEP_RAW and pgbouncer pool vars in .env.example. --- .env.example | 10 +++++ deploy/docker-compose.yml | 38 +++++++++++++++++++ .../migrations/000003_add_updated_at.down.sql | 3 ++ .../migrations/000003_add_updated_at.up.sql | 15 ++++++++ deploy/nginx/nginx.conf | 15 +++++++- deploy/postgres/init.sql | 7 +++- 6 files changed, 85 insertions(+), 3 deletions(-) create mode 100644 deploy/migrations/000003_add_updated_at.down.sql create mode 100644 deploy/migrations/000003_add_updated_at.up.sql diff --git a/.env.example b/.env.example index 414ab1d..6723d69 100644 --- a/.env.example +++ b/.env.example @@ -34,9 +34,19 @@ FFMPEG_PRESET=veryfast ENCODE_MODE=cbr FFMPEG_HWACCEL=auto TRANSCODER_PIPE_INPUT=true +# Phase 15 cost: delete raw/{id}/original.mp4 after ready unless true; lifecycle 7d via mc ilm is safety net +KEEP_RAW=false +# Phase 15: set DATABASE_URL to pgbouncer to enable pooled connections (transaction mode, statement_cache_size=0) +# DATABASE_URL=postgres://flowix:flowix@pgbouncer:6432/flowix?sslmode=disable +DATABASE_POOL_SIZE=5 +DATABASE_MAX_OVERFLOW=10 +PGBOUNCER_ENABLED=false # Services ports (infra) POSTGRES_PORT=5432 +PGBOUNCER_PORT=6432 +PGBOUNCER_DEFAULT_POOL_SIZE=20 +PGBOUNCER_MAX_CLIENT_CONN=100 MINIO_API_PORT=9000 MINIO_CONSOLE_PORT=9001 RABBITMQ_PORT=5672 diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index ef842f9..224baa1 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -16,6 +16,35 @@ services: interval: 5s retries: 10 + # Phase 15 cost: PgBouncer in transaction mode to reduce Postgres connections (pool_size tuned in services/auth & metadata) + pgbouncer: + image: edoburu/pgbouncer:1.23.1 + environment: + DB_HOST: postgres + DB_PORT: 5432 + DB_USER: ${POSTGRES_USER:-flowix} + DB_PASSWORD: ${POSTGRES_PASSWORD:-flowix} + DB_NAME: ${POSTGRES_DB:-flowix} + POOL_MODE: transaction + MAX_CLIENT_CONN: ${PGBOUNCER_MAX_CLIENT_CONN:-100} + DEFAULT_POOL_SIZE: ${PGBOUNCER_DEFAULT_POOL_SIZE:-20} + RESERVE_POOL_SIZE: 5 + RESERVE_POOL_TIMEOUT: 2 + ADMIN_USERS: ${POSTGRES_USER:-flowix} + AUTH_TYPE: md5 + SERVER_RESET_QUERY: DISCARD ALL + IGNORE_STARTUP_PARAMETERS: extra_float_digits,search_path + ports: + - "${PGBOUNCER_PORT:-6432}:5432" + depends_on: + postgres: + condition: service_healthy + healthcheck: + test: ["CMD-SHELL", "psql -h postgres -p 5432 -U ${POSTGRES_USER:-flowix} -d ${POSTGRES_DB:-flowix} -c 'SELECT 1' >/dev/null 2>&1 || pg_isready -h postgres -p 5432 -U ${POSTGRES_USER:-flowix} -d ${POSTGRES_DB:-flowix}"] + interval: 10s + retries: 10 + start_period: 10s + migrate: image: migrate/migrate:v4.18.2 depends_on: @@ -57,6 +86,8 @@ services: /usr/bin/mc anonymous set download local/$${VIDEO_STORAGE_BUCKET:-videos}/thumbnails || true && /usr/bin/mc anonymous set private local/$${VIDEO_STORAGE_BUCKET:-videos}/raw || true && echo '[minio-setup] bucket policy: private + renditions/thumbnails download, raw private' && + (/usr/bin/mc ilm rule add local/$${VIDEO_STORAGE_BUCKET:-videos} --expire-days '7' --prefix 'raw/' 2>/dev/null && echo '[minio-setup] ilm raw 7d added (rule)') || (/usr/bin/mc ilm add local/$${VIDEO_STORAGE_BUCKET:-videos} --expiry-days '7' --prefix 'raw/' 2>/dev/null && echo '[minio-setup] ilm raw 7d added (legacy)') || echo '[minio-setup] ilm raw already exists or skipped' && + /usr/bin/mc ilm ls local/$${VIDEO_STORAGE_BUCKET:-videos} 2>/dev/null | head -n 20 || true && exit 0 " environment: @@ -103,6 +134,10 @@ services: build: ../services/auth environment: DATABASE_URL: postgres://flowix:flowix@postgres:5432/flowix?sslmode=disable + # Phase 15: set PGBOUNCER_ENABLED=true to route via pgbouncer:5432 (pooled) + DATABASE_POOL_SIZE: ${DATABASE_POOL_SIZE:-5} + DATABASE_MAX_OVERFLOW: ${DATABASE_MAX_OVERFLOW:-10} + PGBOUNCER_ENABLED: ${PGBOUNCER_ENABLED:-false} RABBITMQ_URL: amqp://flowix:flowix@rabbitmq:5672/ JWT_SECRET: ${JWT_SECRET} MINIO_ENDPOINT: minio:9000 @@ -129,6 +164,8 @@ services: build: ../services/metadata environment: DATABASE_URL: postgres://flowix:flowix@postgres:5432/flowix?sslmode=disable + DATABASE_POOL_SIZE: ${DATABASE_POOL_SIZE:-5} + PGBOUNCER_ENABLED: ${PGBOUNCER_ENABLED:-false} JWT_SECRET: ${JWT_SECRET} INTERNAL_TOKEN: ${INTERNAL_TOKEN:-} MINIO_ENDPOINT: minio:9000 @@ -198,6 +235,7 @@ services: ENCODE_MODE: ${ENCODE_MODE:-cbr} FFMPEG_HWACCEL: ${FFMPEG_HWACCEL:-auto} TRANSCODER_PIPE_INPUT: ${TRANSCODER_PIPE_INPUT:-true} + KEEP_RAW: ${KEEP_RAW:-false} METRICS_PORT: ${TRANSCODER_METRICS_PORT:-8004} ports: - "${TRANSCODER_METRICS_PORT:-8004}:8004" diff --git a/deploy/migrations/000003_add_updated_at.down.sql b/deploy/migrations/000003_add_updated_at.down.sql new file mode 100644 index 0000000..bb52556 --- /dev/null +++ b/deploy/migrations/000003_add_updated_at.down.sql @@ -0,0 +1,3 @@ +DROP TRIGGER IF EXISTS update_videos_updated_at ON videos; +DROP FUNCTION IF EXISTS update_updated_at_column(); +ALTER TABLE videos DROP COLUMN IF EXISTS updated_at; diff --git a/deploy/migrations/000003_add_updated_at.up.sql b/deploy/migrations/000003_add_updated_at.up.sql new file mode 100644 index 0000000..b343445 --- /dev/null +++ b/deploy/migrations/000003_add_updated_at.up.sql @@ -0,0 +1,15 @@ +-- Phase 15 cost/storage: add updated_at for cache/CDN and lifecycle tracking +ALTER TABLE videos ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now(); + +CREATE OR REPLACE FUNCTION update_updated_at_column() +RETURNS TRIGGER AS $$ +BEGIN + NEW.updated_at = now(); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS update_videos_updated_at ON videos; +CREATE TRIGGER update_videos_updated_at + BEFORE UPDATE ON videos + FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); diff --git a/deploy/nginx/nginx.conf b/deploy/nginx/nginx.conf index 2ba107e..8919e42 100644 --- a/deploy/nginx/nginx.conf +++ b/deploy/nginx/nginx.conf @@ -36,9 +36,9 @@ http { return 200 "ok\n"; } - # nginx-vod fetches the mapping at /mapping/hls/ for an HLS URL. + # nginx-vod fetches the mapping at /mapping/hls/ (or dash) for HLS/DASH. # Phase 9: INTERNAL_TOKEN is injected via envsubst at container start (see Dockerfile). - location ~ "^/mapping/(?:hls/)?(?[0-9a-fA-F-]{36})$" { + location ~ "^/mapping/(?:hls/|dash/)?(?[0-9a-fA-F-]{36})$" { internal; proxy_pass http://metadata:8002/internal/videos/$video_id/vod; proxy_set_header Host metadata; @@ -62,5 +62,16 @@ http { expires 1d; add_header Cache-Control "public, max-age=86400"; } + + # Phase 15 cost: DASH manifests via same mapped MP4s — nginx-vod supports vod_dash (optional, not yet used by FE) + location ~ "^/dash/[0-9a-fA-F-]{36}/" { + vod dash; + vod_mode mapped; + vod_upstream_location /mapping; + vod_remote_upstream_location /minio; + + expires 1d; + add_header Cache-Control "public, max-age=86400"; + } } } diff --git a/deploy/postgres/init.sql b/deploy/postgres/init.sql index 51db9d3..c2953c6 100644 --- a/deploy/postgres/init.sql +++ b/deploy/postgres/init.sql @@ -29,11 +29,16 @@ CREATE TABLE IF NOT EXISTS videos ( status video_status NOT NULL DEFAULT 'uploaded', thumbnail_s3_key TEXT, visibility video_visibility NOT NULL DEFAULT 'public', - created_at TIMESTAMPTZ NOT NULL DEFAULT now() + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); -- thumbnail for VideoCard preview (added Phase 7 fix) ALTER TABLE videos ADD COLUMN IF NOT EXISTS thumbnail_s3_key TEXT; ALTER TABLE videos ADD COLUMN IF NOT EXISTS visibility video_visibility NOT NULL DEFAULT 'public'; +ALTER TABLE videos ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now(); +CREATE OR REPLACE FUNCTION update_updated_at_column() RETURNS TRIGGER AS $$ BEGIN NEW.updated_at = now(); RETURN NEW; END; $$ LANGUAGE plpgsql; +DROP TRIGGER IF EXISTS update_videos_updated_at ON videos; +CREATE TRIGGER update_videos_updated_at BEFORE UPDATE ON videos FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); CREATE TABLE IF NOT EXISTS video_renditions ( video_id UUID NOT NULL REFERENCES videos(id) ON DELETE CASCADE, From 0c2bf72874903eb1ab69627f4db4b09d5f700fec Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 22:49:13 +0300 Subject: [PATCH 2/7] feat(transcoder): delete raw after ready unless KEEP_RAW --- services/transcoder/app/consumer.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/services/transcoder/app/consumer.py b/services/transcoder/app/consumer.py index d17a528..bff701c 100644 --- a/services/transcoder/app/consumer.py +++ b/services/transcoder/app/consumer.py @@ -40,6 +40,7 @@ BUCKET = os.getenv("VIDEO_STORAGE_BUCKET", "videos") METADATA_URL = os.getenv("METADATA_URL", "http://metadata:8002") INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "") +KEEP_RAW = os.getenv("KEEP_RAW", "false").lower() in ("1", "true", "yes") QUEUE = "video.uploaded" DLX_EXCHANGE = "dlx" DLQ = QUEUE + ".dlq" @@ -76,7 +77,9 @@ def get_minio(): ) -def update_status(video_id: str, status: str, renditions=None, thumbnail_s3_key: str | None = None): +def update_status( + video_id: str, status: str, renditions=None, thumbnail_s3_key: str | None = None +) -> bool: url = f"{METADATA_URL}/internal/videos/{video_id}/status" payload: dict = {"status": status} if renditions: @@ -90,8 +93,10 @@ def update_status(video_id: str, status: str, renditions=None, thumbnail_s3_key: r = requests.patch(url, json=payload, headers=headers, timeout=5) r.raise_for_status() log.info("updated %s -> %s", video_id, status) + return True except Exception as e: log.error("metadata update %s failed: %s", video_id, e) + return False def probe_video(path: str) -> dict | None: @@ -824,7 +829,14 @@ def process_message(body: bytes): log.warning("thumbnail upload failed: %s", e) thumb_key = None - update_status(video_id, "ready", renditions, thumb_key) + ok = update_status(video_id, "ready", renditions, thumb_key) + # Phase 15 cost: delete raw only after metadata confirmed ready. Non-fatal; lifecycle 7d is safety net. + if ok and not KEEP_RAW: + try: + mc.remove_object(BUCKET, s3_key) + log.info("cleaned raw s3://%s/%s after ready (KEEP_RAW=false)", BUCKET, s3_key) + except Exception as e: + log.warning("raw cleanup failed for %s: %s", s3_key, e) _shutdown = False From 47e68ac2d7806a2d8bd799569b2ad358a60f118f Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 22:49:13 +0300 Subject: [PATCH 3/7] feat(auth): tune pool for pgbouncer and add keep_raw support --- services/auth/src/core/db.py | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/services/auth/src/core/db.py b/services/auth/src/core/db.py index 3309f7b..ec94d10 100644 --- a/services/auth/src/core/db.py +++ b/services/auth/src/core/db.py @@ -1,3 +1,5 @@ +import os + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from .config import settings @@ -23,7 +25,36 @@ def _async_url(url: str) -> str: return url -engine = create_async_engine(_async_url(settings.database_url), echo=False, pool_pre_ping=True) +def _resolve_url(raw: str) -> str: + # Phase 15: allow PgBouncer via PGBOUNCER_ENABLED without changing DATABASE_URL in .env + if os.getenv("PGBOUNCER_ENABLED", "").lower() in ("1", "true", "yes"): + # rewrite postgres host -> pgbouncer service (inside docker network port 5432) + if "pgbouncer" not in raw: + raw = raw.replace("@postgres:", "@pgbouncer:").replace("@postgres/", "@pgbouncer/") + raw = raw.replace("@localhost:", "@pgbouncer:").replace("localhost:", "pgbouncer:") + return raw + + +def _pool_kwargs(url: str) -> dict: + # Phase 15: tune pool for PgBouncer (transaction mode requires statement_cache_size=0) + is_pgbouncer = "pgbouncer" in url or os.getenv("PGBOUNCER_ENABLED", "").lower() in ( + "1", + "true", + "yes", + ) + kwargs: dict = { + "echo": False, + "pool_pre_ping": True, + "pool_size": int(os.getenv("DATABASE_POOL_SIZE", "5")), + "max_overflow": int(os.getenv("DATABASE_MAX_OVERFLOW", "10")), + } + if is_pgbouncer: + kwargs["connect_args"] = {"statement_cache_size": 0} + return kwargs + + +_resolved_url = _resolve_url(settings.database_url) +engine = create_async_engine(_async_url(_resolved_url), **_pool_kwargs(_resolved_url)) SessionLocal = async_sessionmaker(engine, expire_on_commit=False) From 5d5354539ac83658042ccb017f7f6b3e391fbf52 Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 22:49:13 +0300 Subject: [PATCH 4/7] feat(metadata): add updated_at and pgbouncer pool tuning --- services/metadata/cmd/server/main.go | 29 ++++++++++++++++++- services/metadata/internal/model/video.go | 1 + .../metadata/internal/repository/video.go | 12 ++++---- 3 files changed, 35 insertions(+), 7 deletions(-) diff --git a/services/metadata/cmd/server/main.go b/services/metadata/cmd/server/main.go index 94ba4dd..cdbdff9 100644 --- a/services/metadata/cmd/server/main.go +++ b/services/metadata/cmd/server/main.go @@ -27,6 +27,7 @@ import ( "flowix/metadata/internal/storage" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/rs/zerolog" zlog "github.com/rs/zerolog/log" @@ -44,6 +45,14 @@ func main() { } // pgx doesn't like sslmode, strip it for compat (same as auth) dbURL = stripSSLMode(dbURL) + // Phase 15: if PGBOUNCER_ENABLED, rewrite host to pgbouncer service (transaction pooling) + if strings.ToLower(os.Getenv("PGBOUNCER_ENABLED")) == "true" && !strings.Contains(dbURL, "pgbouncer") { + dbURL = strings.ReplaceAll(dbURL, "@postgres:", "@pgbouncer:") + dbURL = strings.ReplaceAll(dbURL, "@postgres/", "@pgbouncer/") + dbURL = strings.ReplaceAll(dbURL, "@localhost:", "@pgbouncer:") + dbURL = strings.ReplaceAll(dbURL, "localhost:", "pgbouncer:") + dbURL = strings.ReplaceAll(dbURL, "127.0.0.1:", "pgbouncer:") + } jwtSecret := os.Getenv("JWT_SECRET") if jwtSecret == "" { jwtSecret = "change-me-super-secret-jwt-key-32chars" @@ -73,7 +82,25 @@ func main() { } logger := zerolog.New(os.Stdout).With().Timestamp().Logger() - pool, err := pgxpool.New(context.Background(), dbURL) + // Phase 15: tune pool for PgBouncer (transaction mode requires simple protocol, no prepared statements) + poolCfg, err := pgxpool.ParseConfig(dbURL) + if err != nil { + log.Fatalf("pgxpool parse: %v", err) + } + if v := os.Getenv("DATABASE_POOL_SIZE"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + poolCfg.MaxConns = int32(n) + } + } else { + // default 5 conns when behind PgBouncer, else pgx default + if strings.Contains(dbURL, "pgbouncer") { + poolCfg.MaxConns = 5 + } + } + if strings.Contains(dbURL, "pgbouncer") || strings.ToLower(os.Getenv("PGBOUNCER_ENABLED")) == "true" { + poolCfg.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol + } + pool, err := pgxpool.NewWithConfig(context.Background(), poolCfg) if err != nil { log.Fatalf("pgxpool: %v", err) } diff --git a/services/metadata/internal/model/video.go b/services/metadata/internal/model/video.go index beb4ad7..b5d50ad 100644 --- a/services/metadata/internal/model/video.go +++ b/services/metadata/internal/model/video.go @@ -35,6 +35,7 @@ type Video struct { ThumbnailS3Key *string `json:"thumbnail_s3_key,omitempty"` ThumbnailURL *string `json:"thumbnail_url,omitempty"` CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` Renditions []Rendition `json:"renditions,omitempty"` } diff --git a/services/metadata/internal/repository/video.go b/services/metadata/internal/repository/video.go index 7003a0f..02887f0 100644 --- a/services/metadata/internal/repository/video.go +++ b/services/metadata/internal/repository/video.go @@ -19,9 +19,9 @@ func NewVideoRepo(pool *pgxpool.Pool) *VideoRepo { return &VideoRepo{pool: pool} func (r *VideoRepo) Create(ctx context.Context, ownerID, title, description string) (*model.Video, error) { id := uuid.New().String() vis := model.VisibilityPublic - q := `INSERT INTO videos (id, owner_id, title, description, status, visibility) VALUES ($1,$2,$3,$4,'uploaded',$5) RETURNING id, owner_id, title, description, duration, status, visibility, thumbnail_s3_key, created_at` + q := `INSERT INTO videos (id, owner_id, title, description, status, visibility) VALUES ($1,$2,$3,$4,'uploaded',$5) RETURNING id, owner_id, title, description, duration, status, visibility, thumbnail_s3_key, created_at, updated_at` v := &model.Video{} - err := r.pool.QueryRow(ctx, q, id, ownerID, title, description, vis).Scan(&v.ID, &v.OwnerID, &v.Title, &v.Description, &v.Duration, &v.Status, &v.Visibility, &v.ThumbnailS3Key, &v.CreatedAt) + err := r.pool.QueryRow(ctx, q, id, ownerID, title, description, vis).Scan(&v.ID, &v.OwnerID, &v.Title, &v.Description, &v.Duration, &v.Status, &v.Visibility, &v.ThumbnailS3Key, &v.CreatedAt, &v.UpdatedAt) if err != nil { return nil, fmt.Errorf("create video: %w", err) } @@ -35,9 +35,9 @@ func (r *VideoRepo) Create(ctx context.Context, ownerID, title, description stri } func (r *VideoRepo) GetByID(ctx context.Context, id string) (*model.Video, error) { - q := `SELECT v.id, v.owner_id, u.email, v.title, v.description, v.duration, v.status, COALESCE(v.visibility::text,'public'), v.thumbnail_s3_key, v.created_at FROM videos v LEFT JOIN users u ON u.id=v.owner_id WHERE v.id=$1` + q := `SELECT v.id, v.owner_id, u.email, v.title, v.description, v.duration, v.status, COALESCE(v.visibility::text,'public'), v.thumbnail_s3_key, v.created_at, v.updated_at FROM videos v LEFT JOIN users u ON u.id=v.owner_id WHERE v.id=$1` v := &model.Video{} - err := r.pool.QueryRow(ctx, q, id).Scan(&v.ID, &v.OwnerID, &v.OwnerEmail, &v.Title, &v.Description, &v.Duration, &v.Status, &v.Visibility, &v.ThumbnailS3Key, &v.CreatedAt) + err := r.pool.QueryRow(ctx, q, id).Scan(&v.ID, &v.OwnerID, &v.OwnerEmail, &v.Title, &v.Description, &v.Duration, &v.Status, &v.Visibility, &v.ThumbnailS3Key, &v.CreatedAt, &v.UpdatedAt) if err != nil { return nil, err } @@ -62,7 +62,7 @@ func (r *VideoRepo) GetByID(ctx context.Context, id string) (*model.Video, error } func (r *VideoRepo) List(ctx context.Context, limit, offset int) ([]model.Video, error) { - q := `SELECT v.id, v.owner_id, u.email, v.title, v.description, v.duration, v.status, COALESCE(v.visibility::text,'public'), v.thumbnail_s3_key, v.created_at FROM videos v LEFT JOIN users u ON u.id=v.owner_id ORDER BY v.created_at DESC LIMIT $1 OFFSET $2` + q := `SELECT v.id, v.owner_id, u.email, v.title, v.description, v.duration, v.status, COALESCE(v.visibility::text,'public'), v.thumbnail_s3_key, v.created_at, v.updated_at FROM videos v LEFT JOIN users u ON u.id=v.owner_id ORDER BY v.created_at DESC LIMIT $1 OFFSET $2` rows, err := r.pool.Query(ctx, q, limit, offset) if err != nil { return nil, err @@ -71,7 +71,7 @@ func (r *VideoRepo) List(ctx context.Context, limit, offset int) ([]model.Video, var out []model.Video for rows.Next() { var v model.Video - if err := rows.Scan(&v.ID, &v.OwnerID, &v.OwnerEmail, &v.Title, &v.Description, &v.Duration, &v.Status, &v.Visibility, &v.ThumbnailS3Key, &v.CreatedAt); err == nil { + if err := rows.Scan(&v.ID, &v.OwnerID, &v.OwnerEmail, &v.Title, &v.Description, &v.Duration, &v.Status, &v.Visibility, &v.ThumbnailS3Key, &v.CreatedAt, &v.UpdatedAt); err == nil { if v.ThumbnailS3Key != nil { u := "/thumbnails/" + v.ID + "/thumb.jpg" v.ThumbnailURL = &u From ea58196b807c16c14372c1765b07aed0d006ffd5 Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 22:49:13 +0300 Subject: [PATCH 5/7] test(transcoder): add keep_raw lifecycle tests --- services/transcoder/tests/test_consumer.py | 109 ++++++++++++++++++++- 1 file changed, 106 insertions(+), 3 deletions(-) diff --git a/services/transcoder/tests/test_consumer.py b/services/transcoder/tests/test_consumer.py index b707e2c..ec46662 100644 --- a/services/transcoder/tests/test_consumer.py +++ b/services/transcoder/tests/test_consumer.py @@ -235,7 +235,9 @@ def test_process_message_idempotent_failed_skip(): def test_declare_topology(): mock_ch = MagicMock() cons.declare_topology(mock_ch) - mock_ch.exchange_declare.assert_called_once_with(exchange=cons.DLX_EXCHANGE, exchange_type="direct", durable=True) + mock_ch.exchange_declare.assert_called_once_with( + exchange=cons.DLX_EXCHANGE, exchange_type="direct", durable=True + ) # dlq, retry, main + 3 fan-out queues (Phase 12) assert mock_ch.queue_declare.call_count == 6 calls = [c[1].get("queue") for c in mock_ch.queue_declare.call_args_list] @@ -245,10 +247,14 @@ def test_declare_topology(): for fq in cons.FANOUT_QUEUES: assert fq in calls # retry queue has TTL - retry_call = [c for c in mock_ch.queue_declare.call_args_list if c[1].get("queue") == cons.RETRY_QUEUE][0] + retry_call = [ + c for c in mock_ch.queue_declare.call_args_list if c[1].get("queue") == cons.RETRY_QUEUE + ][0] assert retry_call[1]["arguments"]["x-message-ttl"] == cons.RETRY_TTL_MS # main queue has DLX - main_call = [c for c in mock_ch.queue_declare.call_args_list if c[1].get("queue") == cons.QUEUE][0] + main_call = [ + c for c in mock_ch.queue_declare.call_args_list if c[1].get("queue") == cons.QUEUE + ][0] assert main_call[1]["arguments"]["x-dead-letter-exchange"] == cons.DLX_EXCHANGE @@ -256,3 +262,100 @@ def test_fps_from_probe_preserves_and_caps(): assert cons._fps_from_probe({"streams": [{"avg_frame_rate": "24/1"}]}) == 24 assert cons._fps_from_probe({"streams": [{"avg_frame_rate": "60/1"}]}) == 30 assert cons._fps_from_probe(None) == 30 + + +def test_process_message_keep_raw_false_deletes(monkeypatch): + body = json.dumps( + {"video_id": "vid-keep-false", "s3_key": "raw/vid-keep-false/original.mp4"} + ).encode() + fake_minio = MagicMock() + fake_minio.stat_object.return_value = MagicMock(size=100) + fake_minio.fget_object.return_value = None + fake_minio.fput_object.return_value = None + monkeypatch.setattr(cons, "KEEP_RAW", False) + with ( + patch("app.consumer.get_minio", return_value=fake_minio), + patch("app.consumer.update_status", return_value=True) as mock_status, + patch("app.consumer.probe_video", return_value={}), + patch("app.consumer.encode_audio", return_value=None), + patch("app.consumer.transcode_one", return_value=None), + patch("app.consumer.transcode_thumbnail", return_value=None), + patch("app.consumer.transcode_thumbnail_from_rendition", return_value=None), + patch("os.path.exists", return_value=False), + ): + cons.process_message(body) + fake_minio.remove_object.assert_called_once_with( + cons.BUCKET, "raw/vid-keep-false/original.mp4" + ) + assert mock_status.call_args_list[-1][0][1] == "ready" + + +def test_process_message_keep_raw_true_skips_delete(monkeypatch): + body = json.dumps( + {"video_id": "vid-keep-true", "s3_key": "raw/vid-keep-true/original.mp4"} + ).encode() + fake_minio = MagicMock() + fake_minio.stat_object.return_value = MagicMock(size=100) + fake_minio.fget_object.return_value = None + fake_minio.fput_object.return_value = None + monkeypatch.setattr(cons, "KEEP_RAW", True) + with ( + patch("app.consumer.get_minio", return_value=fake_minio), + patch("app.consumer.update_status", return_value=True), + patch("app.consumer.probe_video", return_value={}), + patch("app.consumer.encode_audio", return_value=None), + patch("app.consumer.transcode_one", return_value=None), + patch("app.consumer.transcode_thumbnail", return_value=None), + patch("app.consumer.transcode_thumbnail_from_rendition", return_value=None), + patch("os.path.exists", return_value=False), + ): + cons.process_message(body) + fake_minio.remove_object.assert_not_called() + + +def test_process_message_raw_cleanup_failure_not_raises(monkeypatch): + body = json.dumps( + {"video_id": "vid-clean-fail", "s3_key": "raw/vid-clean-fail/original.mp4"} + ).encode() + fake_minio = MagicMock() + fake_minio.stat_object.return_value = MagicMock(size=100) + fake_minio.fget_object.return_value = None + fake_minio.fput_object.return_value = None + fake_minio.remove_object.side_effect = Exception("delete failed") + monkeypatch.setattr(cons, "KEEP_RAW", False) + with ( + patch("app.consumer.get_minio", return_value=fake_minio), + patch("app.consumer.update_status", return_value=True), + patch("app.consumer.probe_video", return_value={}), + patch("app.consumer.encode_audio", return_value=None), + patch("app.consumer.transcode_one", return_value=None), + patch("app.consumer.transcode_thumbnail", return_value=None), + patch("app.consumer.transcode_thumbnail_from_rendition", return_value=None), + patch("os.path.exists", return_value=False), + ): + # should not raise even if remove fails + cons.process_message(body) + fake_minio.remove_object.assert_called_once() + + +def test_process_message_no_delete_when_metadata_update_fails(monkeypatch): + body = json.dumps( + {"video_id": "vid-no-delete", "s3_key": "raw/vid-no-delete/original.mp4"} + ).encode() + fake_minio = MagicMock() + fake_minio.stat_object.return_value = MagicMock(size=100) + fake_minio.fget_object.return_value = None + fake_minio.fput_object.return_value = None + monkeypatch.setattr(cons, "KEEP_RAW", False) + with ( + patch("app.consumer.get_minio", return_value=fake_minio), + patch("app.consumer.update_status", return_value=False), + patch("app.consumer.probe_video", return_value={}), + patch("app.consumer.encode_audio", return_value=None), + patch("app.consumer.transcode_one", return_value=None), + patch("app.consumer.transcode_thumbnail", return_value=None), + patch("app.consumer.transcode_thumbnail_from_rendition", return_value=None), + patch("os.path.exists", return_value=False), + ): + cons.process_message(body) + fake_minio.remove_object.assert_not_called() From 43203b72baaa94cef5cfeef59dbb69164f3dcb13 Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 23:34:59 +0300 Subject: [PATCH 6/7] fix(infra): use pgbouncer latest tag (1.23.1 not on hub) --- deploy/docker-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 224baa1..827bc96 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -18,7 +18,7 @@ services: # Phase 15 cost: PgBouncer in transaction mode to reduce Postgres connections (pool_size tuned in services/auth & metadata) pgbouncer: - image: edoburu/pgbouncer:1.23.1 + image: edoburu/pgbouncer:latest environment: DB_HOST: postgres DB_PORT: 5432 From 9f3cd4c065503eb35ee158d1ed1da0474c442e43 Mon Sep 17 00:00:00 2001 From: SlavaKuntsov Date: Wed, 2 Sep 2026 23:39:44 +0300 Subject: [PATCH 7/7] fix(infra): postgres md5 for pgbouncer scram compat --- deploy/docker-compose.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 827bc96..e73dba0 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -1,10 +1,14 @@ services: postgres: image: postgres:16-alpine + command: postgres -c password_encryption=md5 environment: POSTGRES_USER: ${POSTGRES_USER:-flowix} POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-flowix} POSTGRES_DB: ${POSTGRES_DB:-flowix} + # Phase 15: pgbouncer uses md5, so postgres must store md5 hashes (scram breaks pgbouncer md5) + POSTGRES_INITDB_ARGS: "--auth-host=md5 --auth-local=md5" + POSTGRES_HOST_AUTH_METHOD: md5 ports: - "${POSTGRES_PORT:-5432}:5432" volumes: