diff --git a/crates/titen-api/migrations/017_thread_bundles.sql b/crates/titen-api/migrations/017_thread_bundles.sql new file mode 100644 index 0000000..3264582 --- /dev/null +++ b/crates/titen-api/migrations/017_thread_bundles.sql @@ -0,0 +1,8 @@ +-- 017 — thread bundles (#thread-bundle Phase 2) +-- A bundle is N schedule rows sharing bundle_id. seq orders execution; +-- total is the item count (integrity check). Root post is seq 0. +-- Rows keep their own status/result so partial failures are visible per item. +ALTER TABLE schedules ADD COLUMN bundle_id TEXT; +ALTER TABLE schedules ADD COLUMN bundle_seq INTEGER; +ALTER TABLE schedules ADD COLUMN bundle_total INTEGER; +CREATE INDEX IF NOT EXISTS idx_schedules_bundle ON schedules(bundle_id); diff --git a/crates/titen-api/src/routes/mod.rs b/crates/titen-api/src/routes/mod.rs index 6844752..8c4171b 100644 --- a/crates/titen-api/src/routes/mod.rs +++ b/crates/titen-api/src/routes/mod.rs @@ -9,4 +9,5 @@ pub mod posts; pub mod schedules; pub mod settings; pub mod threads; +pub mod threads_bundle; pub mod trends; diff --git a/crates/titen-api/src/routes/schedules.rs b/crates/titen-api/src/routes/schedules.rs index 49fb745..c751912 100644 --- a/crates/titen-api/src/routes/schedules.rs +++ b/crates/titen-api/src/routes/schedules.rs @@ -394,10 +394,31 @@ pub async fn approve_schedule( ) -> (StatusCode, Json) { // TODO: extract approver identity from auth context when per-user auth is added match state.store.approve_schedule(&id, Some("api")).await { - Ok(schedule) => ( - StatusCode::OK, - Json(serde_json::json!({ "data": schedule })), - ), + Ok(schedule) => { + // Thread-bundle HITL: approving the ROOT approves the whole + // bundle — members 1..n (draft) become pending-waiting so the + // chain runs after the root publishes. Approving a non-root + // item only affects that item. + if let Some(ref bundle_id) = schedule.bundle_id { + if schedule.bundle_seq == Some(0) { + match state + .store + .set_bundle_status(bundle_id, "draft", "bundle_waiting") + .await + { + Ok(n) if n > 0 => tracing::info!( + "Bundle {bundle_id} approved: {n} waiting item(s) queued" + ), + Ok(_) => {} + Err(e) => tracing::error!("Bundle {bundle_id} approve cascade failed: {e}"), + } + } + } + ( + StatusCode::OK, + Json(serde_json::json!({ "data": schedule })), + ) + } Err(e) => { let code = if matches!(e, titen_core::TitenError::ScheduleNotFound(_)) { StatusCode::NOT_FOUND @@ -440,10 +461,38 @@ pub async fn reject_schedule( ) -> (StatusCode, Json) { let reason = body.and_then(|b| b.reason.clone()); match state.store.reject_schedule(&id, reason.as_deref()).await { - Ok(schedule) => ( - StatusCode::OK, - Json(serde_json::json!({ "data": schedule })), - ), + Ok(schedule) => { + // Thread-bundle: rejecting one member strands the rest of the + // chain. The cascade runs only after reject succeeds (reject + // itself fails for non-rejectable states like published), so an + // invalid request cannot destroy scheduled content. + if let Some(ref bundle_id) = schedule.bundle_id { + match state + .store + .fail_remaining_bundle( + bundle_id, + &format!( + "bundle chain rejected at seq {}: {}", + schedule.bundle_seq.unwrap_or(0), + reason.as_deref().unwrap_or("no reason") + ), + ) + .await + { + Ok(n) if n > 0 => { + tracing::info!("Bundle {bundle_id} rejected: failed {n} remaining item(s)"); + } + Ok(_) => {} + Err(e) => { + tracing::error!("Bundle {bundle_id} cascade fail failed: {e}") + } + } + } + ( + StatusCode::OK, + Json(serde_json::json!({ "data": schedule })), + ) + } Err(e) => { let code = if matches!(e, titen_core::TitenError::ScheduleNotFound(_)) { StatusCode::NOT_FOUND diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs new file mode 100644 index 0000000..f145a78 --- /dev/null +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -0,0 +1,301 @@ +//! Thread bundles (#thread-bundle Phase 2): one root post plus chained +//! replies, publishable instantly or as a single scheduled slot. + +use axum::{Json, extract::State, http::StatusCode}; +use uuid::Uuid; + +/// Validate a single bundle item's media configuration (cheap, pre-insert). +fn validate_item( + item: &titen_core::models::BundleItem, +) -> Result<(), (StatusCode, Json)> { + let media_type = item.media_type.as_deref().unwrap_or("TEXT"); + let urls: Vec = item.media_urls.clone().unwrap_or_default(); + match media_type { + "TEXT" => { + if item.caption.as_deref().unwrap_or("").trim().is_empty() { + return Err(bad("TEXT items require a caption", "INVALID_ITEM")); + } + } + "IMAGE" => { + if urls.len() != 1 { + return Err(bad( + "IMAGE items require exactly 1 media_urls entry", + "INVALID_ITEM", + )); + } + } + "VIDEO" => { + if urls.len() != 1 { + return Err(bad( + "VIDEO items require exactly 1 media_urls entry", + "INVALID_ITEM", + )); + } + } + "CAROUSEL" => { + if !(2..=20).contains(&urls.len()) { + return Err(bad( + "CAROUSEL items require 2-20 media_urls entries", + "INVALID_ITEM", + )); + } + } + other => { + return Err(bad( + &format!("Unsupported media_type: {other}"), + "INVALID_ITEM", + )); + } + } + // Reply targets must reference an earlier item or an external post. + if let Some(serde_json::Value::Number(n)) = &item.reply_to { + let idx = n.as_i64().unwrap_or(i64::MIN); + if idx < 0 { + return Err(bad("reply_to index must be >= 0", "INVALID_REPLY_TO")); + } + } + Ok(()) +} + +fn bad(message: &str, code: &str) -> (StatusCode, Json) { + ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ "error": message, "code": code })), + ) +} + +/// POST /api/threads — create a thread bundle. +/// +/// Without `scheduled_at`: every item is created `pending` with the same +/// "now"-ish timestamp; the scheduler publishes the root, then the chain +/// hooks promote each next item as the previous one publishes. +/// With `scheduled_at`: all items share that slot; root is `pending` +/// (or `draft` unless auto_approve), the rest wait as `bundle_waiting`. +pub async fn create_thread_bundle( + State(state): State, + Json(input): Json, +) -> (StatusCode, Json) { + use titen_core::time::parse_utc; + + let n_items = input.posts.len(); + if n_items == 0 { + return bad("posts must contain at least 1 item", "EMPTY_BUNDLE"); + } + if n_items > 20 { + return bad("posts is limited to 20 items", "TOO_MANY_ITEMS"); + } + // Index-based reply targets must point at earlier items. + for (i, item) in input.posts.iter().enumerate() { + if let Some(serde_json::Value::Number(num)) = &item.reply_to { + let idx = num.as_i64().unwrap_or(i64::MIN); + if idx >= i as i64 { + return bad( + &format!("item {i}: reply_to index {idx} must reference an earlier item"), + "INVALID_REPLY_TO", + ); + } + } + if let Err(resp) = validate_item(item) { + return resp; + } + // caption length: reuse the 499-char guard + if let Some(ref c) = item.caption { + if c.chars().count() > 499 { + return bad( + &format!( + "item {i}: caption exceeds the 499-character limit (got {})", + c.chars().count() + ), + "CAPTION_TOO_LONG", + ); + } + } + // alt_text: not plumbed through the schedule path yet — reject + // explicitly instead of silently discarding accessibility text. + if item + .alt_text + .as_deref() + .map(str::trim) + .map(str::len) + .unwrap_or(0) + > 0 + { + return bad( + &format!("item {i}: alt_text is not yet supported on bundles; omit it for now"), + "ALT_TEXT_UNSUPPORTED", + ); + } + } + + // Resolve mode + timestamp + let (scheduled_at, root_status) = match input.scheduled_at.as_deref() { + Some(sa) => { + if parse_utc(sa).is_none() { + return bad( + &format!( + "scheduled_at must be an ISO 8601 / RFC 3339 timestamp with offset, got: {sa}" + ), + "INVALID_SCHEDULED_AT", + ); + } + ( + sa.to_string(), + if input.auto_approve { + "pending" + } else { + "draft" + }, + ) + } + None => { + // Instant mode: everything is pending immediately. Give each item + // its slot = now (root) — chain promotion handles the order. + (chrono::Utc::now().to_rfc3339(), "pending") + } + }; + + let bundle_id = format!("bundle_{}", Uuid::now_v7()); + + // Single transaction: either the whole bundle exists or nothing does. + // This kills every partial-state class (orphaned drafts, gaps in seq, + // waiting rows stranded by a mid-loop failure). + let mut tx = match state.store.begin().await { + Ok(tx) => tx, + Err(e) => { + return bad( + &format!("failed to begin transaction: {e}"), + "CREATE_FAILED", + ); + } + }; + + let mut created = Vec::with_capacity(n_items); + + for (i, item) in input.posts.iter().enumerate() { + // Reply target: index -> defer via bundle marker (the referenced + // item's real post id only exists after it publishes). External + // string ids pass through as-is. + let reply_to_id = match &item.reply_to { + Some(serde_json::Value::Number(num)) => { + let seq = num.as_i64().unwrap_or(i64::MIN); + Some(format!("bundle:{bundle_id}:{seq}")) + } + Some(serde_json::Value::String(s)) if !s.trim().is_empty() => { + Some(s.trim().to_string()) + } + _ => None, + }; + + let status = if i == 0 { + root_status.to_string() + } else { + // Non-root items wait until their predecessor publishes. In + // instant mode the root is pending, so the next tick publishes + // it and chain-promotes item 1. + "bundle_waiting".to_string() + }; + + let row = titen_core::models::CreateBundleItem { + account_id: input.account_id.clone(), + media_type: item + .media_type + .clone() + .unwrap_or_else(|| "TEXT".to_string()), + caption: item.caption.clone(), + media_urls: item.media_urls.clone(), + scheduled_at: scheduled_at.clone(), + reply_to_id, + bundle_id: bundle_id.clone(), + bundle_seq: i as i64, + bundle_total: n_items as i64, + status: status.clone(), + }; + + let id = Uuid::now_v7().to_string(); + let create_result = state.store.create_bundle_item_tx(&mut tx, &id, &row).await; + match create_result { + Ok(schedule) => created.push(serde_json::json!({ + "id": schedule.id, + "seq": i, + "status": schedule.status, + })), + Err(e) => { + // tx drops here → rollback: nothing partial survives. + return bad( + &format!("failed to create item {i}: {e} — bundle rolled back"), + "CREATE_FAILED", + ); + } + } + } + + if let Err(e) = tx.commit().await { + return bad(&format!("failed to commit bundle: {e}"), "CREATE_FAILED"); + } + + let status = StatusCode::CREATED; + + ( + status, + Json(serde_json::json!({ + "data": { + "bundle_id": bundle_id, + "items": created, + "mode": if input.scheduled_at.is_some() { "scheduled" } else { "instant" }, + } + })), + ) +} + +/// GET /api/threads/{bundle_id} — bundle overview (items in seq order). +pub async fn get_thread_bundle( + State(state): State, + axum::extract::Path(bundle_id): axum::extract::Path, +) -> (StatusCode, Json) { + match state.store.get_bundle_schedules(&bundle_id).await { + Ok(items) if items.is_empty() => ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "bundle not found", "code": "BUNDLE_NOT_FOUND" })), + ), + Ok(items) => { + let published = state + .store + .count_bundle_status(&bundle_id, "published") + .await + .unwrap_or(0); + let failed = state + .store + .count_bundle_status(&bundle_id, "failed") + .await + .unwrap_or(0); + let total = items.len() as i64; + let bundle_status = if failed > 0 { + "failed" + } else if published == total { + "published" + } else if published > 0 { + "partial" + } else { + "queued" + }; + ( + StatusCode::OK, + Json(serde_json::json!({ + "data": { + "bundle_id": bundle_id, + "status": bundle_status, + "published": published, + "failed": failed, + "total": total, + "items": items, + } + })), + ) + } + // DB error is a server fault — 500, not 404. + Err(e) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": e.to_string(), "code": "INTERNAL" })), + ), + } +} diff --git a/crates/titen-api/src/server.rs b/crates/titen-api/src/server.rs index 02dcd81..0e8589d 100644 --- a/crates/titen-api/src/server.rs +++ b/crates/titen-api/src/server.rs @@ -381,6 +381,14 @@ pub async fn serve( get(routes::posts::get_post).delete(routes::posts::delete_post), ) .route("/api/posts/{id}/insights", get(routes::posts::get_insights)) + .route( + "/api/threads", + post(routes::threads_bundle::create_thread_bundle), + ) + .route( + "/api/threads/{bundle_id}", + get(routes::threads_bundle::get_thread_bundle), + ) .route( "/api/schedules", get(routes::schedules::list_schedules).post(routes::schedules::create_schedule), diff --git a/crates/titen-api/tests/api_schedules.rs b/crates/titen-api/tests/api_schedules.rs index a5a481e..324671c 100644 --- a/crates/titen-api/tests/api_schedules.rs +++ b/crates/titen-api/tests/api_schedules.rs @@ -368,6 +368,125 @@ async fn list_schedules_pagination_meta() { // ─── #232: schedulable thread replies (reply_to_id) ───────────────────── +#[tokio::test] +async fn create_thread_bundle_creates_pending_root_and_waiting_replies() { + let pool = test_pool().await; + let state = test_state(pool.clone()); + let app = test_app(state); + + let account = create_test_account(&app, &pool).await; + let account_id = account["id"].as_str().unwrap(); + + let req = axum::http::Request::builder() + .method("POST") + .uri("/api/threads") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&json!({ + "account_id": account_id, + "scheduled_at": "2099-06-15T12:00:00Z", + "auto_approve": true, + "posts": [ + { "media_type": "TEXT", "caption": "root post" }, + { "media_type": "TEXT", "caption": "reply 1", "reply_to": 0 }, + { "media_type": "IMAGE", "caption": "reply 2 with media", + "media_urls": ["https://s3.ajianaz.dev/example/image.jpg"], + "reply_to": 0 } + ] + })) + .unwrap(), + )) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 201); + + let body = body_to_json(resp).await; + let bundle_id = body["data"]["bundle_id"].as_str().unwrap().to_string(); + let items = body["data"]["items"].as_array().unwrap(); + assert_eq!(items.len(), 3); + assert_eq!(items[0]["status"], "pending"); + assert_eq!(items[1]["status"], "bundle_waiting"); + assert_eq!(items[2]["status"], "bundle_waiting"); + + // GET the bundle back + let req = axum::http::Request::builder() + .method("GET") + .uri(format!("/api/threads/{bundle_id}")) + .body(Body::empty()) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 200); + let body = body_to_json(resp).await; + assert_eq!(body["data"]["total"], 3); + assert_eq!(body["data"]["status"], "queued"); + assert_eq!(body["data"]["items"].as_array().unwrap().len(), 3); +} + +#[tokio::test] +async fn create_thread_bundle_rejects_reply_to_forward_index() { + let pool = test_pool().await; + let state = test_state(pool.clone()); + let app = test_app(state); + + let account = create_test_account(&app, &pool).await; + let account_id = account["id"].as_str().unwrap(); + + let req = axum::http::Request::builder() + .method("POST") + .uri("/api/threads") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&json!({ + "account_id": account_id, + "posts": [ + { "media_type": "TEXT", "caption": "root", + "reply_to": 1 } + ] + })) + .unwrap(), + )) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 400); + let body = body_to_json(resp).await; + assert_eq!(body["code"], "INVALID_REPLY_TO"); +} + +#[tokio::test] +async fn create_thread_bundle_instant_mode_all_items_waiting_but_root() { + let pool = test_pool().await; + let state = test_state(pool.clone()); + let app = test_app(state); + + let account = create_test_account(&app, &pool).await; + let account_id = account["id"].as_str().unwrap(); + + // No scheduled_at => instant mode: root publishes on the next tick. + let req = axum::http::Request::builder() + .method("POST") + .uri("/api/threads") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&json!({ + "account_id": account_id, + "posts": [ + { "media_type": "TEXT", "caption": "instant root" }, + { "media_type": "TEXT", "caption": "instant reply", "reply_to": 0 } + ] + })) + .unwrap(), + )) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 201); + + let body = body_to_json(resp).await; + assert_eq!(body["data"]["mode"], "instant"); + let items = body["data"]["items"].as_array().unwrap(); + assert_eq!(items[0]["status"], "pending"); + assert_eq!(items[1]["status"], "bundle_waiting"); +} + #[tokio::test] async fn create_reply_schedule_persists_reply_to_id() { let pool = test_pool().await; diff --git a/crates/titen-api/tests/common/mod.rs b/crates/titen-api/tests/common/mod.rs index c6f91a9..3bc859f 100644 --- a/crates/titen-api/tests/common/mod.rs +++ b/crates/titen-api/tests/common/mod.rs @@ -50,6 +50,14 @@ pub fn test_app(state: AppState) -> Router { "/api/accounts", get(routes::accounts::list_accounts).post(routes::accounts::create_account), ) + .route( + "/api/threads", + post(routes::threads_bundle::create_thread_bundle), + ) + .route( + "/api/threads/{bundle_id}", + get(routes::threads_bundle::get_thread_bundle), + ) .route( "/api/accounts/{id}", put(routes::accounts::update_account).delete(routes::accounts::delete_account), diff --git a/crates/titen-core/src/models.rs b/crates/titen-core/src/models.rs index a0e2971..6df241a 100644 --- a/crates/titen-core/src/models.rs +++ b/crates/titen-core/src/models.rs @@ -127,6 +127,10 @@ pub struct Schedule { /// When set, the scheduler publishes this schedule as a reply to that /// Threads post ID instead of a root-level post (TEXT only). See #232. pub reply_to_id: Option, + /// Thread bundle: rows sharing a bundle_id execute in bundle_seq order. + pub bundle_id: Option, + pub bundle_seq: Option, + pub bundle_total: Option, /// Free-form origin identifier ("cmo-agent", "ci-pipeline", ...) set via /// the ingest endpoint (#242). NULL for dashboard-created schedules. pub source: Option, @@ -153,6 +157,58 @@ pub struct CreateSchedule { pub reply_to_id: Option, } +/// Internal bundle row creator (Phase 2). Not serde-deserializable from any +/// request body: only /api/threads constructs it, so clients can never +/// inject bundle membership into普通 schedules. +#[derive(Debug)] +pub struct CreateBundleItem { + pub account_id: String, + pub media_type: String, + pub caption: Option, + pub media_urls: Option>, + pub scheduled_at: String, + pub reply_to_id: Option, + pub bundle_id: String, + pub bundle_seq: i64, + pub bundle_total: i64, + pub status: String, +} + +/// One entry inside a thread bundle (`POST /api/threads`). +#[derive(Debug, Deserialize, ToSchema)] +pub struct BundleItem { + /// `TEXT`, `IMAGE`, `VIDEO`, or `CAROUSEL`. Defaults to TEXT. + #[serde(default)] + pub media_type: Option, + #[serde(default)] + pub caption: Option, + /// IMAGE/VIDEO: single URL. CAROUSEL: 2–20 URLs. + #[serde(default)] + pub media_urls: Option>, + #[serde(default)] + pub alt_text: Option, + /// Reply target: an integer = the 0-based index of a previous item in + /// this bundle (0 = root post); a string = an existing Threads post ID. + #[serde(default)] + pub reply_to: Option, +} + +/// `POST /api/threads` — a thread bundle: one root post plus optional +/// chained replies. Without `scheduled_at` everything publishes immediately; +/// with it, one schedule slot executes the whole chain in order. +#[derive(Debug, Deserialize, ToSchema)] +pub struct CreateThreadBundle { + pub account_id: String, + /// ISO 8601 / RFC 3339 timestamp. Optional — absent = publish now. + #[serde(default)] + pub scheduled_at: Option, + /// Skip HITL review (schedule mode only). Default: false. + #[serde(default)] + pub auto_approve: bool, + /// 1..=20 items; item 0 is the root post. + pub posts: Vec, +} + /// Request body for the agent/CI ingest endpoint (#242). /// /// Deliberately has NO `auto_approve` field: ingested drafts NEVER skip the diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index aaa438b..f75c458 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -125,6 +125,20 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< } } + // Thread-bundle Phase 2: promote bundle members whose ROOT item (seq 0) + // is due. Members 1..n enter as 'bundle_waiting' and only become + // 'pending' once the previous item publishes — this keeps the chain + // ordered while everything shares the existing schedules machinery. + if let Err(e) = store.promote_due_bundle_roots().await { + error!("Failed to promote due bundle roots: {e}"); + } + // Reconcile: promote any waiting member whose predecessor published, + // regardless of how publication happened (covers lost promote hooks, + // hook errors, and crashes between publish and promote). + if let Err(e) = store.reconcile_bundle_promotions().await { + error!("Failed to reconcile bundle promotions: {e}"); + } + let due_schedules = match store.get_due_schedules().await { Ok(s) => s, Err(e) => { @@ -169,6 +183,7 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< Some(&format!("Account not found: {e}")), ) .await; + cascade_fail_bundle(store, &schedule, &format!("account not found: {e}")).await; continue; } }; @@ -182,6 +197,7 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< let _ = store .update_schedule_status(&schedule.id, "failed", None, Some("Account is inactive")) .await; + cascade_fail_bundle(store, &schedule, "account is inactive").await; continue; } @@ -260,11 +276,49 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< schedule.location_id.as_deref(), schedule.reply_to_id.as_deref(), ); - let reply_to = req + let mut reply_to = req .reply_to_id .take() .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); + // Bundle chain markers ("bundle::") resolve to the + // real Threads post id of the referenced bundle member. + if let Some(marker) = reply_to.clone() { + if marker.starts_with("bundle:") { + match resolve_bundle_marker(store, &marker).await { + Ok(Some(Some(target_id))) => { + reply_to = Some(target_id); + } + // Target failed or not yet published / malformed marker: + // fail this item cleanly instead of sending a literal + // marker to Threads. + Ok(Some(None)) | Ok(None) => { + // Ok(None): malformed marker — permanent problem, + // fail the item instead of publishing garbage. + let msg = format!("bundle reply target unavailable: {marker}"); + let _ = store + .update_schedule_status(&schedule.id, "failed", None, Some(&msg)) + .await; + cascade_fail_bundle(store, &schedule, &msg).await; + continue; + } + Err(e) => { + // Transient store error (lock timeout, pool + // exhaustion): NOT fatal — reset to pending so the + // next tick retries the lookup instead of destroying + // publishable content. + warn!( + "Bundle marker lookup errored for schedule {} — retrying next tick: {e}", + schedule.id + ); + let _ = store + .update_schedule_status(&schedule.id, "pending", None, None) + .await; + continue; + } + } + } + } req.reply_to_id = reply_to.clone(); let result = match crate::publisher::publish(client, &account, &req).await { @@ -324,6 +378,18 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< .update_schedule_status(&schedule.id, "published", Some(&result_str), None) .await; + // Thread-bundle: promote the next waiting member of this bundle. + if let Some(ref bundle_id) = schedule.bundle_id { + let seq = schedule.bundle_seq.unwrap_or(0); + match store.promote_next_bundle_item(bundle_id, seq).await { + Ok(n) if n > 0 => { + info!("Bundle {bundle_id}: promoted {n} item(s) after seq {seq}"); + } + Ok(_) => {} + Err(e) => error!("Bundle {bundle_id} chain advance failed: {e}"), + } + } + info!( "Schedule {} published as post {} for @{}", schedule.id, post_id, account.username @@ -333,6 +399,28 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< let _ = store .update_schedule_status(&schedule.id, "failed", None, Some(&e)) .await; + // Thread-bundle: a failed chain item strands the rest — fail + // every remaining waiting member so nothing hangs forever. + if let Some(ref bundle_id) = schedule.bundle_id { + match store + .fail_remaining_bundle( + bundle_id, + &format!( + "bundle chain stopped at seq {}: {e}", + schedule.bundle_seq.unwrap_or(0) + ), + ) + .await + { + Ok(n) if n > 0 => { + warn!( + "Bundle {bundle_id}: failed {n} remaining item(s) after chain break" + ); + } + Ok(_) => {} + Err(e2) => error!("Bundle {bundle_id} cascade fail failed: {e2}"), + } + } error!("Schedule {} failed: {e}", schedule.id); } } @@ -340,3 +428,63 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< Ok(()) } + +/// Fail every remaining waiting member of a bundle when one of its members +/// fails outside the publish-result path (account missing/inactive, HITL +/// reject, etc.). No-op for non-bundle schedules. +async fn cascade_fail_bundle(store: &Store, schedule: &crate::models::Schedule, reason: &str) { + if let Some(ref bundle_id) = schedule.bundle_id { + match store + .fail_remaining_bundle(bundle_id, &format!("bundle chain stopped: {reason}")) + .await + { + Ok(n) if n > 0 => { + warn!("Bundle {bundle_id}: failed {n} remaining item(s) — {reason}"); + } + Ok(_) => {} + Err(e) => error!("Bundle {bundle_id} cascade fail failed: {e}"), + } + } +} + +/// Resolve a `bundle::` reply marker to the referenced +/// member's published Threads post id. +/// Returns: +/// - `Ok(Some(Some(post_id)))` — referenced member published; proceed. +/// - `Ok(Some(None))` — referenced member failed/cancelled; its reply can +/// never exist, so the caller should fail this item permanently. +/// - `Ok(None)` — malformed marker; permanent, fail the item. +/// - `Err(e)` — transient store error; the caller should retry next tick +/// instead of destroying scheduled content. +async fn resolve_bundle_marker( + store: &Store, + marker: &str, +) -> crate::Result>> { + // Malformed markers and missing members are PERMANENT (client-supplied + // input) -> Ok(None): the caller fails the item. Only store errors are + // Err: transient, the caller retries next tick. + let rest = match marker.strip_prefix("bundle:") { + Some(r) => r, + None => return Ok(None), // unreachable — caller pre-gates on prefix + }; + let (bundle_id, seq_str) = match rest.rsplit_once(':') { + Some(pair) => pair, + None => return Ok(None), + }; + let seq: i64 = match seq_str.parse() { + Ok(s) => s, + Err(_) => return Ok(None), + }; + let members = store.get_bundle_schedules(bundle_id).await?; + let target = match members.iter().find(|s| s.bundle_seq == Some(seq)) { + Some(t) => t, + // Referenced member does not exist (deleted bundle, bad index): + // permanent — its post id can never appear. + None => return Ok(None), + }; + match target.status.as_str() { + "published" => Ok(Some(target.result_post_id.clone())), + "failed" | "cancelled" | "rejected" => Ok(Some(None)), + _ => Ok(Some(None)), + } +} diff --git a/crates/titen-core/src/store.rs b/crates/titen-core/src/store.rs index fc8263c..811bbff 100644 --- a/crates/titen-core/src/store.rs +++ b/crates/titen-core/src/store.rs @@ -442,6 +442,21 @@ impl Store { } } + // 017 — thread bundles: bundle_id/bundle_seq/bundle_total columns + for stmt in split_sql_statements(include_str!( + "../../titen-api/migrations/017_thread_bundles.sql" + )) { + let result = sqlx::query(&stmt).execute(&self.pool).await; + if let Err(e) = result { + let msg = e.to_string(); + if !msg.contains("duplicate column") && !msg.contains("already exists") { + return Err(TitenError::DatabaseError(format!( + "migration 017 failed: {msg}" + ))); + } + } + } + Ok(()) } @@ -1012,6 +1027,161 @@ impl Store { self.get_schedule(id).await } + /// Begin a transaction (thread-bundle atomic creation). + pub async fn begin(&self) -> Result> { + self.pool.begin().await.map_err(Into::into) + } + + /// Create one thread-bundle member inside `tx` (Phase 2). Bundle + /// membership is internal-only: CreateBundleItem is not serde-compatible + /// with any request body, so clients cannot inject it. + pub async fn create_bundle_item_tx( + &self, + tx: &mut sqlx::SqliteConnection, + id: &str, + input: &crate::models::CreateBundleItem, + ) -> Result { + let media_urls = input + .media_urls + .as_ref() + .map(|urls| serde_json::to_string(urls).unwrap_or_default()); + let caption = input + .caption + .as_deref() + .map(crate::models::sanitize_caption); + + sqlx::query( + "INSERT INTO schedules (id, account_id, media_type, caption, text_attachment, media_urls, scheduled_at, status, location_id, reply_to_id, bundle_id, bundle_seq, bundle_total) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(id) + .bind(&input.account_id) + .bind(&input.media_type) + .bind(&caption) + .bind(None::) // text_attachment + .bind(&media_urls) + .bind(&input.scheduled_at) + .bind(&input.status) + .bind(None::) // location_id + .bind(&input.reply_to_id) + .bind(&input.bundle_id) + .bind(input.bundle_seq) + .bind(input.bundle_total) + .execute(&mut *tx) + .await?; + + sqlx::query_as::<_, Schedule>("SELECT * FROM schedules WHERE id = ?") + .bind(id) + .fetch_one(&mut *tx) + .await + .map_err(|_| TitenError::ScheduleNotFound(id.to_string())) + } + + /// All bundle members of one bundle, ordered by seq (thread-bundle Phase 2). + pub async fn get_bundle_schedules(&self, bundle_id: &str) -> Result> { + sqlx::query_as::<_, Schedule>( + "SELECT * FROM schedules \ + WHERE bundle_id = ? \ + ORDER BY bundle_seq ASC", + ) + .bind(bundle_id) + .fetch_all(&self.pool) + .await + .map_err(Into::into) + } + + /// Bulk status update for a bundle (approve/reject all members at once). + /// Returns the number of rows transitioned draft -> target status. + pub async fn set_bundle_status(&self, bundle_id: &str, from: &str, to: &str) -> Result { + let result = + sqlx::query("UPDATE schedules SET status = ? WHERE bundle_id = ? AND status = ?") + .bind(to) + .bind(bundle_id) + .bind(from) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Count bundle members still pending/published-root for partial tracking. + pub async fn count_bundle_status(&self, bundle_id: &str, status: &str) -> Result { + let row: (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM schedules WHERE bundle_id = ? AND status = ?") + .bind(bundle_id) + .bind(status) + .fetch_one(&self.pool) + .await?; + Ok(row.0) + } + + /// Promote due bundle roots: seq-0 rows of a bundle whose root + /// scheduled_at has passed move bundle_waiting -> pending so the normal + /// due-schedule path claims them. Idempotent. + pub async fn promote_due_bundle_roots(&self) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'pending' \ + WHERE status = 'bundle_waiting' \ + AND bundle_seq = 0 \ + AND datetime(scheduled_at) <= datetime('now')", + ) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Chain advance: when bundle member `seq` publishes, promote the next + /// waiting member(s) whose reply chain resolves to an item at or before + /// `seq`. Member k replies to seq k-1 by design (bundle_reply_to is the + /// *index*); a reply to an external post id (not an index) is promoted + /// immediately after the root publishes. + pub async fn promote_next_bundle_item(&self, bundle_id: &str, seq: i64) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'pending' \ + WHERE bundle_id = ? \ + AND status = 'bundle_waiting' \ + AND bundle_seq = ? + 1", + ) + .bind(bundle_id) + .bind(seq) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Reconcile bundle chains: promote waiting members whose predecessor + /// (seq-1) is published. Self-healing — covers lost promote hooks, + /// hook errors, and crashes between publish and promote. Idempotent. + pub async fn reconcile_bundle_promotions(&self) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'pending' \ + WHERE status = 'bundle_waiting' \ + AND bundle_seq > 0 \ + AND EXISTS ( \ + SELECT 1 FROM schedules prev \ + WHERE prev.bundle_id = schedules.bundle_id \ + AND prev.bundle_seq = schedules.bundle_seq - 1 \ + AND prev.status = 'published' \ + )", + ) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Fail every remaining waiting member of a bundle (root failed — the + /// chain can never start). + pub async fn fail_remaining_bundle(&self, bundle_id: &str, error: &str) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'failed', error = ? \ + WHERE bundle_id = ? AND status = 'bundle_waiting'", + ) + .bind(error) + .bind(bundle_id) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + pub async fn get_schedule(&self, id: &str) -> Result { sqlx::query_as::<_, Schedule>("SELECT * FROM schedules WHERE id = ?") .bind(id)