//! Audio/video streaming: scan-status gating, subscription + bundle access,
//! response shape, play counters.
//!
//! Complements `storage.rs` (which covers the core access matrix:
//! free vs paid, anonymous vs authenticated, creator preview, draft
//! visibility) and `video.rs` (video-specific happy paths). This file covers:
//!
//!   - scan_status variations (Pending, Quarantined, HeldForReview)
//!     gate on creator identity
//!   - Subscription-based access on paid items
//!   - Bundle parent grants access to child items
//!   - Item missing audio_s3_key / video_s3_key returns 404
//!   - Response shape: `stream_url` + `expires_in` fields
//!   - `play_count` increments on stream
//!   - `unique_play` tracking for authenticated viewers

use crate::harness::TestHarness;
use serde_json::Value;

/// Create a creator with a published audio item. Returns
/// (user_id, project_id, item_id, s3_key). The InMemoryStorage is
/// seeded so `presign_download` returns a usable URL.
async fn setup_audio_item(
    h: &mut TestHarness,
    price_cents: i64,
) -> (String, String, String, String) {
    let setup = h
        .create_creator_with_item("streamer", "audio", price_cents)
        .await;
    let s3_key = format!("test/{}/audio/track.mp3", setup.item_id);
    sqlx::query(
        "UPDATE items SET audio_s3_key = $1, scan_status = 'clean', is_public = true \
         WHERE id = $2::uuid",
    )
    .bind(&s3_key)
    .bind(&setup.item_id)
    .execute(&h.db)
    .await
    .unwrap();
    sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
        .bind(&setup.project_id)
        .execute(&h.db)
        .await
        .unwrap();
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"audio data".to_vec());
    (
        setup.user_id.to_string(),
        setup.project_id,
        setup.item_id,
        s3_key,
    )
}

/// The download/stream access gate MUST honor `current_period_end`. An item
/// subscription still flagged `status = 'active'` (its cancel/expiry webhook
/// lapsed or never arrived) but whose paid period has EXPIRED must not grant
/// access. A hand-written `status = 'active' AND paused_at IS NULL` drops the
/// period clause and lets a lapsed subscriber pull the file on the very route
/// that protects content. The gate routes through `subscriptions::has_access`,
/// which enforces the clause in one sealed place.
/// If anyone re-inlines the predicate without the period clause, the first
/// assertion (lapsed → 403) flips to 200 and this test fails.
#[tokio::test]
async fn stream_denies_item_subscription_with_lapsed_period() {
    let mut h = TestHarness::with_storage().await;
    // Paid audio item owned by the creator (price > 0 ⇒ access is gated).
    let (_creator, _project, item_id, _s3) = setup_audio_item(&mut h, 500).await;
    let item_uuid: uuid::Uuid = item_id.parse().unwrap();

    // Item-level subscription tier (project_id NULL, item_id set).
    let tier_id: uuid::Uuid = sqlx::query_scalar(
        "INSERT INTO subscription_tiers (project_id, item_id, name, price_cents) \
         VALUES (NULL, $1, 'Item Tier', 500) RETURNING id",
    )
    .bind(item_uuid)
    .fetch_one(&h.db)
    .await
    .unwrap();

    // A fan with a known password (seeded directly; we only need to log in).
    let hash = makenotwork::auth::hash_password("password123").unwrap();
    let fan_id: makenotwork::db::UserId = sqlx::query_scalar(
        "INSERT INTO users (username, email, password_hash, email_verified) \
         VALUES ('lapsedfan', 'lapsedfan@example.com', $1, true) RETURNING id",
    )
    .bind(&hash)
    .fetch_one(&h.db)
    .await
    .unwrap();

    // Active-status item subscription whose paid period ended a day ago.
    sqlx::query(
        "INSERT INTO subscriptions \
         (subscriber_id, tier_id, item_id, project_id, stripe_subscription_id, \
          stripe_customer_id, status, paused_at, current_period_end) \
         VALUES ($1, $2, $3, NULL, 'sub_lapsed', 'cus_lapsed', 'active', NULL, NOW() - INTERVAL '1 day')",
    )
    .bind(fan_id)
    .bind(tier_id)
    .bind(item_uuid)
    .execute(&h.db)
    .await
    .unwrap();

    // Act as the fan (the creator is logged in from setup_audio_item).
    h.client.post_form("/logout", "").await;
    h.login("lapsedfan", "password123").await;

    // Lapsed period → denied on the download gate.
    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status.as_u16(),
        403,
        "lapsed-period subscriber must be denied the download gate, got {} {}",
        resp.status,
        resp.text
    );

    // Same subscriber, period pushed into the future → now passes.
    sqlx::query(
        "UPDATE subscriptions SET current_period_end = NOW() + INTERVAL '30 days' \
         WHERE subscriber_id = $1",
    )
    .bind(fan_id)
    .execute(&h.db)
    .await
    .unwrap();

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status, 200,
        "subscriber within the paid period must pass the gate, got {} {}",
        resp.status, resp.text
    );
}

#[tokio::test]
async fn stream_url_response_has_expected_shape() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(resp.status, 200, "{} {}", resp.status, resp.text);

    let data: Value = resp.json();
    assert!(
        data["stream_url"].is_string(),
        "Response must contain stream_url"
    );
    assert!(
        data["expires_in"].is_u64(),
        "Response must contain numeric expires_in"
    );
    // Test storage backend returns http://test-storage/<key>.
    assert!(
        data["stream_url"]
            .as_str()
            .unwrap()
            .contains("audio/track.mp3"),
        "URL should reference the seeded key"
    );
}

#[tokio::test]
async fn stream_url_404_when_item_has_no_audio_key() {
    let mut h = TestHarness::with_storage().await;
    let setup = h.create_creator_with_item("noaudio", "audio", 0).await;
    // Mark the item public + clean but DON'T set audio_s3_key, the
    // handler should refuse to mint a streaming URL for a "naked" item.
    sqlx::query("UPDATE items SET is_public = true, scan_status = 'clean' WHERE id = $1::uuid")
        .bind(&setup.item_id)
        .execute(&h.db)
        .await
        .unwrap();
    sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
        .bind(&setup.project_id)
        .execute(&h.db)
        .await
        .unwrap();

    let resp = h
        .client
        .get(&format!("/api/stream/{}", setup.item_id))
        .await;
    assert_eq!(
        resp.status.as_u16(),
        404,
        "Item without audio_s3_key must 404"
    );
}

#[tokio::test]
async fn stream_url_404_for_quarantined_item_to_non_creator() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
    sqlx::query("UPDATE items SET scan_status = 'quarantined' WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();
    // Log out, the creator is currently authenticated.
    h.client.post_form("/logout", "").await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status.as_u16(),
        404,
        "Quarantined items must not stream to non-creators"
    );
}

#[tokio::test]
async fn stream_url_404_for_pending_scan_to_non_creator() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
    sqlx::query("UPDATE items SET scan_status = 'pending' WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();
    h.client.post_form("/logout", "").await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status.as_u16(),
        404,
        "Pending-scan items must not stream to non-creators (fail-closed)"
    );
}

#[tokio::test]
async fn stream_url_creator_can_preview_held_for_review_item() {
    let mut h = TestHarness::with_storage().await;
    let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await;
    sqlx::query("UPDATE items SET scan_status = 'held_for_review' WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();
    // Creator remains logged in from setup.

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status, 200,
        "Creator must be able to preview their own HeldForReview content: {} {}",
        resp.status, resp.text
    );
}

#[tokio::test]
async fn stream_url_authenticated_non_buyer_gets_403_on_paid_item() {
    let mut h = TestHarness::with_storage().await;
    let (_creator, _, item_id, _) = setup_audio_item(&mut h, 999).await;
    // Log out creator; create a different user with no purchase.
    h.client.post_form("/logout", "").await;
    h.signup("randomuser", "random@test.com", "password123")
        .await;
    h.login("randomuser", "password123").await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status.as_u16(),
        403,
        "Authenticated non-buyer on paid item must get 403, not 401: {} {}",
        resp.status,
        resp.text
    );
}

#[tokio::test]
async fn stream_url_increments_play_count() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;

    // Stream twice (free item, anyone can hit it).
    h.client.post_form("/logout", "").await;
    for _ in 0..2 {
        let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
        assert_eq!(resp.status, 200, "{}", resp.text);
    }

    let count: i32 = sqlx::query_scalar("SELECT play_count FROM items WHERE id = $1::uuid")
        .bind(&item_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(count, 2, "play_count should increment per stream call");
}

#[tokio::test]
async fn stream_url_records_unique_play_for_authenticated_user() {
    let mut h = TestHarness::with_storage().await;
    let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await;

    // Switch to a different authenticated user.
    h.client.post_form("/logout", "").await;
    let listener_id = h
        .signup("listener", "listener@test.com", "password123")
        .await;
    h.login("listener", "password123").await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(resp.status, 200, "{}", resp.text);

    let has_play: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM user_plays WHERE user_id = $1 AND item_id = $2::uuid)",
    )
    .bind(listener_id)
    .bind(&item_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(
        has_play,
        "user_plays should record the listener for unique-listener tracking"
    );
}

// `version_download_includes_license_url_when_preset_set` and
// `version_download_omits_license_url_without_preset` were here until
// 2026-08-26. `GET /api/versions/{id}/download` answered JSON carrying a
// `license_url`; it now answers 303 to the presigned URL (`8fc6b1af`, option
// (a)), so a redirect has nowhere to put that field and these two assert a
// shape that no longer exists.
//
// The licence itself is not lost and was never reached through this field:
// `/api/items/{id}/license.txt` is linked directly from `pages/item.html` and
// `pages/library_downloads.html`, fetched by `static/page-item-1.js`, and
// published as a v1 OpenAPI path. Nothing consumed the copy in this response.

#[tokio::test]
async fn stream_url_404_for_nonexistent_item() {
    let mut h = TestHarness::with_storage().await;
    let bogus = "00000000-0000-0000-0000-000000000000";
    let resp = h.client.get(&format!("/api/stream/{bogus}")).await;
    assert_eq!(resp.status.as_u16(), 404);
}

#[tokio::test]
async fn stream_url_expires_in_scales_with_duration() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
    // Set a long duration. The handler computes expiry as
    // `max(duration * 2, 3600)`, so 5000s should give 10000s expiry.
    sqlx::query("UPDATE items SET duration_seconds = 5000 WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(resp.status, 200, "{} {}", resp.status, resp.text);
    let data: Value = resp.json();
    let expires_in = data["expires_in"].as_u64().unwrap();
    assert_eq!(
        expires_in, 10000,
        "expires_in should be 2x duration for long tracks (got {expires_in})"
    );
}