Status Service & Expiry Cleanup

Status Service & Expiry Cleanup

The Status Service implements ephemeral, story-style user statuses in the Rust (axum/sqlx) backend: create, list (with cursor pagination), group-by-user overview, ownership-guarded delete, a 24-hour expiry TTL, and a background purge job that deletes expired rows and their associated media.

Purpose and Scope

This page documents the status module of streaming-backend end to end:

• The HTTP handlers in controller.rs
• The domain logic in service.rs
• The request/response DTOs in models.rs
• The statuses table schema (migrations 004_statuses.up.sql and 009_add_status_bg_color.up.sql)
• The expiry purge background task spawned in main.rs

Sibling capabilities that this page only references (not re-documents) are the shared infrastructure services the status module relies on: QueryService (cursor pagination helpers), MediaCleanupService (object-storage deletion), the snowflake ID generator in AppState, and the AuthUser auth middleware. For those, see the corresponding catalog pages. The React Native / Expo front-end status screens (streaming/src/components/StatusComposer.tsx, streaming/src/components/StatusViewer.tsx) and the web admin UI (streaming-frontend/components/status/*) are out of scope for this Rust-backend page.

Overview

Statuses behave like social “stories”: a user posts a short-lived item (media URL, caption, background color) that is visible to other users only while it has not expired. Key design decisions visible in the source:

• Ephemerality by TTL, not by scheduled deletion alone. Every status row stores expires_at (Unix seconds), computed as now + 86400 at creation time in create_status. All read queries filter expires_at > now, so expired stories are invisible immediately even before the purge job runs.
• Batched, transactionally safe background cleanup. A tokio task spawned in main.rs runs every 600 seconds and calls StatusService::purge_expired_statuses, which deletes up to 50 expired rows per pass inside a transaction using FOR UPDATE SKIP LOCKED, cascading to media deletion.
• Ownership enforcement. Deleting a status requires the authenticated user to be the row owner; otherwise the service returns AppError::Forbidden.
• Distributed-safe IDs. Status IDs come from the AppState snowflake generator, and every i64 ID in the JSON payloads is serialized as a string (via serialize_i64_as_str) to avoid JavaScript number-precision loss in the mobile/web clients.
• Cursor pagination. Status lists are fetched newest-first with an opaque created_at-based cursor, fetched through the shared QueryService::fetch_cursor_page.

Architecture

Component roles:

• Handlers (controller.rs) are thin adapters: they parse query strings, extract the authenticated user, and delegate everything to StatusService. They never touch SQL directly.
• StatusService is a unit struct (no instance state) whose associated functions take &AppState explicitly — all dependencies (DB pool, storage client, snowflake) are injected via AppState, which keeps the service easy to call from both HTTP handlers and the background task.
• QueryService centralizes parameterized SQL execution and cursor-page mechanics so the status module only declares what columns, joins, and predicates to use (CursorQueryOptions).
• MediaCleanupService::delete_media removes the underlying media blob when a status is deleted or purged, preventing orphaned objects.
• The background task in main.rs is the only caller of purge_expired_statuses; it is spawned once at server startup and runs forever on a 600-second interval.

Why this structure

The separation mirrors the rest of the backend: controllers are route-only, services are static function bundles over AppState, and shared SQL/pagination utilities live in services/. This makes the expiry purge callable from any context (HTTP or background) and keeps the status module free of connection-pool plumbing.

Service Layer Deep Dive

StatusService is declared as a unit struct (pub struct StatusService;) with five pub async fn associated functions (service.rs). Each method receives &AppState, which carries the sqlx pool (state.db), the object-storage client (state.storage), and the snowflake generator (state.snowflake).

list_user_groups — per-user story overview

pub async fn list_user_groups(state: &AppState) -> Result<UserStatusGroupResponse> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() as i64;
    let mut args = PgArguments::default();
    let _ = args.add(now);
    let groups: Vec<UserStatusGroup> = QueryService::fetch_all(
        &state.db,
        "SELECT s.user_id, u.username, MAX(s.created_at) as latest_created_at, COUNT(s.id) as story_count FROM statuses s JOIN users u ON s.user_id = u.id WHERE s.expires_at > $1 GROUP BY s.user_id, u.username ORDER BY latest_created_at DESC",
        args
    ).await?;
    Ok(UserStatusGroupResponse { groups })
}

Source: service.rs

This is the “story tray” query: one row per author who currently has at least one unexpired status, with the timestamp of their most recent story and the count of their live stories. Design intent:

• The WHERE s.expires_at > $1 predicate is the single source of truth for visibility — a user with only expired stories simply does not appear in the tray.
• MAX(s.created_at) is used both as a display value and as the ordering key (ORDER BY latest_created_at DESC), so authors with the freshest content float to the top.
• Timestamps are stored as i64 Unix seconds and compared directly against now computed in Rust with SystemTime::now().duration_since(UNIX_EPOCH), falling back to 0 on clock errors (unwrap_or_default()).

list_user_statuses — cursor-paginated feed for one author

pub async fn list_user_statuses(state: &AppState, target_user_id: i64, cursor: Option<i64>, limit: usize) -> Result<StatusListResponse> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() as i64;
    let mut args = PgArguments::default();
    let _ = args.add(now);
    let _ = args.add(target_user_id);

    let opts = CursorQueryOptions {
        select_cols: Some("s.id, s.user_id, u.username, s.media_url, s.caption, s.bg_color, s.expires_at, s.created_at"),
        table_or_join: "statuses s JOIN users u ON s.user_id = u.id",
        where_clause: "s.expires_at > $1 AND s.user_id = $2",
        cursor_cols: vec!["s.created_at"],
        cursor_val: cursor,
        limit: (limit + 1) as i64,
    };

    let rows: Vec<StatusItem> = QueryService::fetch_cursor_page(&state.db, opts, args).await?;
    let has_more = rows.len() > limit;
    let mut items: Vec<StatusItem> = rows.into_iter().take(limit).collect();
    items.reverse();
    let next_cursor = if has_more { items.first().map(|i| i.created_at) } else { None };
    Ok(StatusListResponse { items, next_cursor })
}

Source: service.rs

Key mechanics:

• limit + 1 over-fetch: the service asks the cursor query for one extra row; if more rows come back than requested, has_more is true and the client receives a next_cursor.
• Newest-first ordering: the underlying cursor query pages in ascending created_at order, so the service fetches the oldest rows of the page first and reverses them before returning — the client always sees the freshest statuses at the top.
• Cursor semantics: next_cursor is the created_at of the first (i.e., oldest) returned item when more pages exist. Since expires_at and created_at are both set to now at insert time and statuses never change, created_at is a stable, unique-enough pagination key per author.
• Expiry filter again: the where_clause includes s.expires_at > $1, so the feed never shows stale stories even if purge has not run yet.

create_status — insert with 24-hour TTL

pub async fn create_status(state: &AppState, user_id: i64, req: CreateStatusRequest) -> Result<StatusItem> {
    let id = state.snowflake.next_id();
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() as i64;
    let expires_at = now + 86400;

    let mut user_args = PgArguments::default();
    let _ = user_args.add(user_id);
    let user = QueryService::fetch_optional::<UserRow>(&state.db, "SELECT username FROM users WHERE id = $1", user_args).await?
        .ok_or_else(|| AppError::NotFound("User not found".into()))?;

    let mut ins_args = PgArguments::default();
    let _ = ins_args.add(id); let _ = ins_args.add(user_id);
    let _ = ins_args.add(req.media_url.clone()); let _ = ins_args.add(req.caption.clone());
    let _ = ins_args.add(req.bg_color.clone()); let _ = ins_args.add(expires_at); let _ = ins_args.add(now);

    QueryService::execute(&state.db, "INSERT INTO statuses (id, user_id, media_url, caption, bg_color, expires_at, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)", ins_args).await?;
    Ok(StatusItem { id, user_id, username: user.username, media_url: req.media_url, caption: req.caption, bg_color: req.bg_color, expires_at, created_at: now })
}

Source: service.rs

• expires_at = now + 86400 hard-codes the 24-hour TTL (86,400 seconds) — the constant lives inline in the service rather than in configuration.
• The user existence check (SELECT username FROM users WHERE id = $1) runs first and returns AppError::NotFound("User not found") for a missing author. The username is fetched eagerly so the response StatusItem can include it without a second join.
• The ID is generated client-side-of-the-DB by the snowflake generator, so inserts never depend on RETURNING or sequence allocation.
• bg_color (added by migration 009_add_status_bg_color.up.sql) is stored as a nullable string and echoed back in the response.

delete_status — ownership-guarded delete with media cascade

pub async fn delete_status(state: &AppState, status_id: i64, user_id: i64) -> Result<()> {
    let mut chk_args = PgArguments::default();
    let _ = chk_args.add(status_id);
    let item = QueryService::fetch_optional::<StatusDeleteCheckRow>(&state.db, "SELECT user_id, media_url FROM statuses WHERE id = $1", chk_args).await?
        .ok_or_else(|| AppError::NotFound("Status story not found".into()))?;
    if item.user_id != user_id { return Err(AppError::Forbidden("Only status owner can delete this story".into())); }
    if let Some(media) = item.media_url { let _ = MediaCleanupService::delete_media(&state.storage, &media).await; }
    let mut del_args = PgArguments::default();
    let _ = del_args.add(status_id);
    QueryService::execute(&state.db, "DELETE FROM statuses WHERE id = $1", del_args).await?;
    Ok(())
}

Source: service.rs

Two sqlx::FromRow helper structs back this method: StatusDeleteCheckRow { user_id, media_url } and, for the purge path, StatusExpiredRow { id, media_url } (service.rs). The delete flow is deliberately check-then-act:

1. Existence check → NotFound("Status story not found") if the row does not exist.
2. Ownership check → Forbidden("Only status owner can delete this story") if row.user_id != auth user_id. Authorization happens in the service, not the route, so every caller is protected.
3. Media cleanup → if the row carries media_url, the object-store blob is deleted (errors swallowed with let _).
4. Row delete → a parameterized DELETE removes the row itself.

The check-then-delete is not atomic (a concurrent purge could delete the row between steps 1 and 4); in practice this surfaces only as a rare NotFound/no-op since deletion is idempotent.

purge_expired_statuses — the expiry cleanup job

pub async fn purge_expired_statuses(state: &AppState) -> Result<usize> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() as i64;
    let mut tx = state.db.begin().await?;
    let mut purge_args = PgArguments::default();
    let _ = purge_args.add(now);
    let expired = sqlx::query_as_with::<_, StatusExpiredRow, _>("SELECT id, media_url FROM statuses WHERE expires_at <= $1 ORDER BY expires_at ASC LIMIT 50 FOR UPDATE SKIP LOCKED", purge_args).fetch_all(&mut *tx).await?;

    let mut count = 0;
    for item in expired {
        if let Some(media) = item.media_url { let _ = MediaCleanupService::delete_media(&state.storage, &media).await; }
        let mut del_args = PgArguments::default();
        let _ = del_args.add(item.id);
        QueryService::execute_tx(&mut tx, "DELETE FROM statuses WHERE id = $1", del_args).await?;
        count += 1;
    }
    tx.commit().await?;
    Ok(count)
}

Source: service.rs

This is the heart of “Expiry Cleanup”. Design details worth calling out:

• Transactional batching: the expired-row SELECT and all DELETEs run inside one explicit transaction (state.db.begin()), committed at the end. If any delete fails, the ? propagates the error, the transaction is rolled back on drop, and no partial cleanup is committed.
• FOR UPDATE SKIP LOCKED: the batch query locks the selected rows and skips any rows already locked by another transaction. This makes the job safe to run concurrently — e.g., multiple backend instances behind a load balancer — because two workers can never delete the same row or block each other; each pass takes up to 50 rows nobody else is working on.
• ORDER BY expires_at ASC LIMIT 50: the oldest-expiring statuses are processed first and the batch is capped at 50 rows per invocation, bounding per-pass work and keeping each transaction short.
• Media cascade: each expired row’s media_url is passed to MediaCleanupService::delete_media (best-effort, errors ignored) before its row is deleted — the same orphan-prevention policy used by delete_status.
• Incomplete-pass tolerance: because the purge loop in main.rs runs every 600 seconds, any pass that deletes fewer than all expired rows (more than 50 expired, or a concurrent lock) simply continues on the next tick. The read-path expires_at > $1 filters guarantee expired rows are invisible in the meantime.

The background loop in main.rs

let cleanup_state = state.clone();
tokio::spawn(async move {
    let mut interval = tokio::time::interval(std::time::Duration::from_secs(600));
    loop {
        interval.tick().await;
        let _ = crate::modules::status::service::StatusService::purge_expired_statuses(&cleanup_state).await;
    }
});

Source: main.rs

The task is spawned at startup right after the identity-key backfill and the gRPC server, before the Axum HTTP server binds. Notes:

• state.clone() hands the task its own AppState handle; AppState is cheaply clonable (shared pool/client handles), which is why the service methods take &AppState.
• tokio::time::interval fires the first tick immediately after the first wait of 600 s, then every 600 s thereafter.
• The result is deliberately ignored (let _), so a transient DB failure in one pass never kills the loop; the next tick retries. This matches the “best-effort, read-path-filters-protect-visibility” philosophy.

Data Model & Persistence

The statuses table is defined by migrations 004_statuses.up.sql (initial schema) and 009_add_status_bg_color.up.sql (adds bg_color). The exact column set is confirmed by the INSERT in create_status and the row structs in the service:

Timestamps are stored as plain Unix-seconds integers rather than TIMESTAMPTZ. This keeps every query trivially comparable with SystemTime::now() computed in Rust and avoids timezone conversions, at the cost of losing Postgres-native datetime functions on this column.

Read-path visibility invariant: every public read (list_user_groups, list_user_statuses) enforces expires_at > now. Purge is a storage-reclamation concern, not a visibility concern — even if the background job is delayed or fails repeatedly, expired stories never leak into API responses.

Core Flow

Status lifecycle

Create → read → delete interaction

Expiry purge pass

Usage Examples

Creating a status (service call site)

The handler extracts the authenticated user and delegates; the service performs the TTL computation and user check:

pub async fn create_status(
    auth_user: AuthUser,
    State(state): State<AppState>,
    Json(req): Json<CreateStatusRequest>,
) -> Result<Json<StatusItem>> {
    let item = StatusService::create_status(&state, auth_user.user_id, req).await?;
    Ok(Json(item))
}

Source: controller.rs

Listing with query-string parsing and clamping

pub async fn list_statuses(
    State(state): State<AppState>,
    Query(q): Query<StatusListQuery>,
) -> Result<Json<StatusListResponse>> {
    let cursor = q.cursor.and_then(|c| c.parse::<i64>().ok());
    let limit = q.limit.and_then(|l| l.parse::<usize>().ok()).unwrap_or(10).min(50);
    let target_user = q.user_id.and_then(|u| u.parse::<i64>().ok());

    if let Some(uid) = target_user {
        let res = StatusService::list_user_statuses(&state, uid, cursor, limit).await?;
        Ok(Json(res))
    } else {
        let groups = StatusService::list_user_groups(&state).await?;
        let dummy = StatusListResponse { items: vec![], next_cursor: None };
        Ok(Json(dummy))
    }
}

Source: controller.rs

The limit query parameter defaults to 10 and is clamped to at most 50; malformed cursor/user_id values degrade to None (first page / tray view) instead of erroring. Note the implementation quirk in the no-user_id branch: list_user_groups is invoked but its result is discarded and an empty items list is returned — the groups are served by the dedicated GET /api/statuses/groups endpoint instead (see list_user_groups handler below).

pub async fn list_user_groups(
    State(state): State<AppState>,
) -> Result<Json<UserStatusGroupResponse>> {
    let res = StatusService::list_user_groups(&state).await?;
    Ok(Json(res))
}

Source: controller.rs

Serializing IDs as strings

All public i64 identifiers are string-serialized so JavaScript clients do not lose precision:

fn serialize_i64_as_str<S>(val: &i64, serializer: S) -> Result<S::Ok, S::Error>
where S: Serializer {
    serializer.serialize_str(&val.to_string())
}

#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
pub struct StatusItem {
    #[serde(serialize_with = "serialize_i64_as_str")]
    pub id: i64,
    #[serde(serialize_with = "serialize_i64_as_str")]
    pub user_id: i64,
    pub username: String,
    pub media_url: Option<String>,
    pub caption: Option<String>,
    pub bg_color: Option<String>,
    pub expires_at: i64,
    pub created_at: i64,
}

Source: models.rs

The same serializer is applied to UserStatusGroup.user_id; expires_at and created_at remain plain numbers because they fit safely in a JS Number for the foreseeable future.

API Reference

HTTP endpoints

(The exact route prefixes are wired by router::build_router in the backend’s router module; the handlers above are the status-module entry points.)

StatusService methods

list_user_groups(state: &AppState) -> Result<UserStatusGroupResponse>

Returns one UserStatusGroup per author having at least one unexpired status, ordered by latest_created_at descending. Executes the group-by query through QueryService::fetch_all.

Throws:

• AppError::Database (via ? propagation): query failure.

list_user_statuses(state: &AppState, target_user_id: i64, cursor: Option<i64>, limit: usize) -> Result<StatusListResponse>

Returns limit statuses for target_user_id (unexpired only), newest first. Fetches limit + 1 rows to compute has_more; next_cursor is the created_at of the oldest returned item when more pages exist, else None.

Parameters:

• target_user_id (i64): author whose statuses to list.
• cursor (Option<i64>): created_at-based page cursor.
• limit (usize): page size (the caller clamps it; the service itself does not clamp).

Throws:

• AppError::Database: query failure.

create_status(state: &AppState, user_id: i64, req: CreateStatusRequest) -> Result<StatusItem>

Generates a snowflake id, computes expires_at = now + 86400, verifies the author exists, and inserts the row.

Parameters:

• user_id (i64): authenticated author (from AuthUser).
• req (CreateStatusRequest): media_url: Option<String>, caption: Option<String>, bg_color: Option<String>.

Throws:

• AppError::NotFound("User not found"): no users row for user_id.
• AppError::Database: insert failure.

delete_status(state: &AppState, status_id: i64, user_id: i64) -> Result<()>

Deletes the status after existence and ownership checks; cascades to media deletion when media_url is present.

Throws:

• AppError::NotFound("Status story not found"): no row for status_id.
• AppError::Forbidden("Only status owner can delete this story"): row.user_id != user_id.
• AppError::Database: delete failure.

purge_expired_statuses(state: &AppState) -> Result<usize>

Deletes up to 50 expired rows in one transaction (FOR UPDATE SKIP LOCKED), deleting each row’s media first. Returns the number of purged rows.

Throws:

• AppError::Database: transaction begin/query/commit failure (transaction rolls back).

DTOs (models.rs)

Configuration Options

The status module has no external config file; its tunables are constants and spawn parameters hard-coded in source:

Failure Modes, Edge Cases & Concurrency

• Expired-but-not-purged visibility: guaranteed invisible because every read filters expires_at > now. Purge lag only affects storage usage, never correctness.
• Purge worker errors are swallowed: let _ = ...purge_expired_statuses(...) in main.rs means a failed pass is silently retried next tick; there is no alerting or backoff. If the DB is down for many cycles, expired rows accumulate until the job succeeds.
• Concurrent purge workers: FOR UPDATE SKIP LOCKED lets multiple instances run the job without double-deleting; a worker only locks rows nobody else holds. With many expired rows (>50 per pass), each tick handles the next batch, so a large backlog drains over several ticks.
• Delete-vs-purge race: delete_status’s check-then-delete is non-atomic; a concurrent purge can remove the row between the SELECT and DELETE, which surfaces as a benign NotFound (already-deleted outcome). The converse (purge deleting a row while a user tries to delete it) is also safe: the second deleter sees NotFound.
• Media deletion is best-effort: delete_media errors are discarded in both delete_status and purge_expired_statuses, so a storage outage can leave orphaned blobs that are never retried by this module.
• limit clamp protects the DB: the handler clamps limit to [1..50]; an attacker passing limit=1000000 gets 50. The service-level limit + 1 over-fetch is bounded by this.
• Snowflake clock assumptions: create_status relies on state.snowflake.next_id(); no failure path is handled here for clock skew — the generator’s guarantees are its own concern.
• SystemTime fallback: duration_since(UNIX_EPOCH).unwrap_or_default() silently uses epoch 0 on clock errors, which would make all statuses appear expired; in practice system clocks behind Unix epoch are not a realistic deployment scenario.

Performance & Operational Considerations

• Index-friendly queries: the hot read predicates are expires_at > $1 (both read paths) and user_id = $2 plus ORDER BY created_at (per-author feed). An index on statuses(expires_at) and a composite index on (user_id, created_at) are what the query shapes assume; the FOR UPDATE ... ORDER BY expires_at purge scan likewise benefits from (expires_at). The exact index set is defined in the migration files (004_statuses.up.sql, 009_add_status_bg_color.up.sql).
• Purge cost per pass is bounded: LIMIT 50 keeps each transaction small (≤50 row locks + ≤50 media deletions), so the 600-second job does not create long-running transactions or lock storms.
• Reads never block on cleanup: because visibility is enforced by the expires_at > now predicate, the read path does not contend with the purge transaction; only the rows the purge locked (FOR UPDATE) are briefly unavailable to concurrent DELETEs, and SKIP LOCKED prevents waits.
• Over-fetch by one row (limit + 1) is the standard cost of cursor pagination with a has_more flag; with a 50-row cap the extra row is negligible.
• Group query cost: list_user_groups aggregates over all unexpired statuses; with a large number of concurrent authors this is the most expensive query in the module and relies on the expires_at index for the filter before the GROUP BY.
• No caching layer: status data is read fresh from Postgres on every request; no in-memory or Redis cache exists in this module (the SQL is parameterized, so the plan cache still applies).
• Error swallowing in background paths: both the purge loop (main.rs) and media deletions ignore errors, so operational visibility depends on logs/APM rather than on the module raising alerts.

Extension Points

• Configurable TTL: expires_at = now + 86400 in create_status is the single place to change the story lifetime (e.g., per-tier TTLs, REPLY-style durations). A future refactor could move 86400 into AppConfig and add per-request override support to CreateStatusRequest.
• CursorQueryOptions reuse: list_user_statuses builds the paginated query declaratively via QueryService::fetch_cursor_page; adding filters (e.g., caption IS NOT NULL, visibility scopes) is a matter of extending where_clause/cursor_cols.
• New status kinds: CreateStatusRequest already carries optional media_url, caption, and bg_color; adding fields is a schema migration plus a DTO field — no change to the service flow.
• Purge policy tuning: batch size (LIMIT 50) and interval (600s) are the two knobs for cleanup cadence; moving them to configuration would allow ops to tune without a rebuild.
• Retry for media cleanup: delete_media failures are currently dropped; a dead-letter queue or a second pass that re-scans rows whose media still exists would make orphan cleanup eventually consistent.

Tests

No dedicated test files for the status module were found in the repository scan (streaming-backend tests, if any, live outside the module tree). The behavior documented above is therefore derived from the production code paths themselves: the read-path expiry filters, the ownership guard, the limit + 1 pagination, and the FOR UPDATE SKIP LOCKED purge transaction are all verifiable in service.rs. When adding tests, the highest-value targets are: (1) list_user_statuses pagination boundaries (empty page, exact page, page+1), (2) purge_expired_statuses with >50 expired rows across two calls, (3) ownership rejection in delete_status, and (4) expired-row invisibility without a purge run.

Related Links

• Status module — controller — HTTP handlers for the endpoints above
• Status module — service — domain logic, TTL, purge implementation
• Status module — models — DTOs and the serialize_i64_as_str helper
• Server entry point — expiry cleanup background task
• Migration 004 — statuses table — initial statuses schema
• Migration 009 — bg_color — adds bg_color column
• For the shared pagination/SQL helpers used here, see the Query Service catalog page; for object-storage deletion, see Media Cleanup Service; for the auth extractor, see Auth Middleware; for the mobile client screens, see the front-end status components (streaming/src/components/StatusComposer.tsx, streaming/src/components/StatusViewer.tsx).