feat(phase-15): storage cost — raw lifecycle, pgbouncer, updated_at, dash

.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

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:
@@ -16,6 +20,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:latest
+    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 +90,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 +138,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 +168,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 +239,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"

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;

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();

deploy/nginx/nginx.conf

@@ -36,9 +36,9 @@ http {
       return 200 "ok\n";
     }
 
-    # nginx-vod fetches the mapping at /mapping/hls/<video_id> for an HLS URL.
+    # nginx-vod fetches the mapping at /mapping/hls/<video_id> (or dash) for HLS/DASH.
     # Phase 9: INTERNAL_TOKEN is injected via envsubst at container start (see Dockerfile).
-    location ~ "^/mapping/(?:hls/)?(?<video_id>[0-9a-fA-F-]{36})$" {
+    location ~ "^/mapping/(?:hls/|dash/)?(?<video_id>[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";
+    }
   }
 }

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,

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)
 
 

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)
 	}

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"`
 }
 

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

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