gonzalo_ticket/

ingest.rs

//! The ingest engine: pull a [`TicketSource`] and persist each [`Ticket`] as a
//! `Record`, using optimistic concurrency (ADR 0005). Re-sync is idempotent —
//! unchanged tickets (same body hash) are skipped, so a full board re-scan is
//! cheap. Depends only on the trait + `Store`, never on a concrete connector.

use crate::{TicketSource, record_key};
use gonzalo_core::{ContentHash, Identity, Meta, PutResult, Record, Revision, Store};
use gonzalo_domain::{RecordCodec, Ticket};

/// How many tickets a sync created, updated, or left untouched.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct IngestSummary {
    pub imported: usize,
    pub updated: usize,
    pub unchanged: usize,
}

/// Failures during ingest.
#[derive(Debug, thiserror::Error)]
pub enum IngestError {
    #[error("ticket source: {0}")]
    Source(#[from] crate::SourceError),
    #[error("store: {0}")]
    Store(#[from] gonzalo_core::CoreError),
    /// Boxed because inlining a `RecordKey` and two revisions made every
    /// `Result<_, IngestError>` in this module 136 bytes wide on its error
    /// path (clippy 1.99's `result_large_err`). `PutResult::Conflict` already
    /// hands back a `Box<Conflict>`, so carrying it through costs nothing and
    /// gives callers the whole conflicting record rather than its revision
    /// alone. The message still prints only the revision — a `Record`'s
    /// `Debug` would include its body.
    #[error("write conflict on {}: expected {:?}, store has {:?}", .0.key, .0.expected, .0.current.revision)]
    Conflict(Box<gonzalo_core::Conflict>),
}

/// Pull all changed tickets from `source` and upsert them into `store`,
/// attributing writes to `author`.
///
/// `scope` is the board/connection discriminator woven into each record key
/// (see [`crate::record_key`]): pass the connection name for board sources so
/// the same external item on two boards lands in two distinct records (#159),
/// or `None` for a single-board / board-agnostic source.
pub async fn ingest(
    source: &dyn TicketSource,
    store: &dyn Store,
    author: &str,
    scope: Option<&str>,
) -> Result<IngestSummary, IngestError> {
    let mut summary = IngestSummary::default();
    let mut cursor = crate::Cursor::default();
    loop {
        let page = source.fetch_changed(&cursor).await?;
        for ticket in &page.tickets {
            match upsert(store, ticket, author, scope).await? {
                Outcome::Imported => summary.imported += 1,
                Outcome::Updated => summary.updated += 1,
                Outcome::Unchanged => summary.unchanged += 1,
            }
        }
        if page.next.0.is_none() || page.next == cursor {
            break;
        }
        cursor = page.next;
    }
    Ok(summary)
}

/// Per-ticket result of an upsert, tallied into [`IngestSummary`].
enum Outcome {
    Imported,
    Updated,
    Unchanged,
}

async fn upsert(
    store: &dyn Store,
    ticket: &Ticket,
    author: &str,
    scope: Option<&str>,
) -> Result<Outcome, IngestError> {
    let key = record_key(ticket, scope);
    let body = ticket.to_body()?;
    let new_hash = ContentHash::of(body.bytes());

    let existing = store.get(&key).await?;
    if let Some(rec) = &existing
        && rec.revision.hash == new_hash
    {
        return Ok(Outcome::Unchanged);
    }

    let expected: Option<Revision> = existing.as_ref().map(|r| r.revision.clone());
    let meta = Meta::new(Identity::new(author), "ticket-ingest");
    let record = match &existing {
        Some(current) => current.update(body, meta),
        None => Record::create(key.clone(), Ticket::KIND, body, meta),
    };
    match store.put(record, expected).await? {
        PutResult::Committed(_) => Ok(if existing.is_some() {
            Outcome::Updated
        } else {
            Outcome::Imported
        }),
        PutResult::Conflict(c) => Err(IngestError::Conflict(c)),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::InMemorySource;
    use gonzalo_domain::{BodyFormat, Provider, State, StateCategory, TicketBody};
    use gonzalo_store_fs::FsStore;
    use std::collections::BTreeMap;

    fn ticket(uid: &str, title: &str) -> Ticket {
        Ticket {
            provider: Provider::GitHub,
            uid: uid.into(),
            display: "#1".into(),
            item_type: "issue".into(),
            title: title.into(),
            state: State {
                category: StateCategory::Open,
                resolution: None,
                raw_name: "Todo".into(),
                raw_id: None,
            },
            priority: None,
            actors: vec![],
            labels: vec![],
            containers: vec![],
            links: vec![],
            body: TicketBody {
                markdown: String::new(),
                format: BodyFormat::Markdown,
                raw: None,
            },
            fields: BTreeMap::new(),
        }
    }

    #[tokio::test]
    async fn imports_then_is_idempotent_then_updates() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsStore::new(dir.path());

        let src = InMemorySource::new(vec![ticket("a", "A"), ticket("b", "B")]);
        let s1 = ingest(&src, &store, "tester", None).await.unwrap();
        assert_eq!(
            s1,
            IngestSummary {
                imported: 2,
                updated: 0,
                unchanged: 0
            }
        );

        let s2 = ingest(&src, &store, "tester", None).await.unwrap();
        assert_eq!(
            s2,
            IngestSummary {
                imported: 0,
                updated: 0,
                unchanged: 2
            }
        );

        let src2 = InMemorySource::new(vec![ticket("a", "A2"), ticket("b", "B")]);
        let s3 = ingest(&src2, &store, "tester", None).await.unwrap();
        assert_eq!(
            s3,
            IngestSummary {
                imported: 0,
                updated: 1,
                unchanged: 1
            }
        );
    }

    /// A source that returns two pages, driving the pagination loop across the
    /// cursor boundary. Stateless: it branches on the incoming cursor value.
    struct PagedSource;

    #[async_trait::async_trait]
    impl crate::TicketSource for PagedSource {
        fn capabilities(&self) -> crate::Capabilities {
            crate::Capabilities::default()
        }

        async fn fetch_changed(&self, cursor: &crate::Cursor) -> crate::Result<crate::Page> {
            match cursor.0.as_deref() {
                None => Ok(crate::Page {
                    tickets: vec![ticket("p1", "P1")],
                    next: crate::Cursor(Some("p2".into())),
                }),
                Some("p2") => Ok(crate::Page {
                    tickets: vec![ticket("p2", "P2")],
                    next: crate::Cursor::default(),
                }),
                Some(other) => Err(crate::SourceError::Backend(format!(
                    "unexpected cursor: {other}"
                ))),
            }
        }

        async fn get(&self, uid: &str) -> crate::Result<Ticket> {
            Ok(ticket(uid, uid))
        }
    }

    #[tokio::test]
    async fn paginates_across_multiple_pages() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsStore::new(dir.path());

        let summary = ingest(&PagedSource, &store, "tester", None).await.unwrap();
        assert_eq!(summary.imported, 2);
    }

    #[tokio::test]
    async fn same_issue_on_two_boards_does_not_thrash_one_record() {
        // #159: the same uid imported under two connections must yield two
        // distinct records (one per board), not alternating overwrites of one.
        let dir = tempfile::tempdir().unwrap();
        let store = FsStore::new(dir.path());

        // Board A puts the card in "Todo"; board B in "Done" — same uid.
        let mut on_b = ticket("caliban-ai/gonzalo#15", "shared");
        on_b.state.raw_name = "Done".into();
        on_b.state.category = StateCategory::Done;
        let src_a = InMemorySource::new(vec![ticket("caliban-ai/gonzalo#15", "shared")]);
        let src_b = InMemorySource::new(vec![on_b]);

        // First import on each board is a fresh record — no collision.
        let a1 = ingest(&src_a, &store, "tester", Some("board-a"))
            .await
            .unwrap();
        let b1 = ingest(&src_b, &store, "tester", Some("board-b"))
            .await
            .unwrap();
        assert_eq!(a1.imported, 1);
        assert_eq!(b1.imported, 1);

        // Re-syncing each is idempotent: neither disturbs the other's record,
        // which is exactly the thrash the shared key used to cause.
        let a2 = ingest(&src_a, &store, "tester", Some("board-a"))
            .await
            .unwrap();
        let b2 = ingest(&src_b, &store, "tester", Some("board-b"))
            .await
            .unwrap();
        assert_eq!(a2.unchanged, 1, "board-a stayed put: {a2:?}");
        assert_eq!(b2.unchanged, 1, "board-b stayed put: {b2:?}");

        // Both records exist side by side under distinct keys.
        let ka = record_key(&ticket("caliban-ai/gonzalo#15", "shared"), Some("board-a"));
        let kb = record_key(&ticket("caliban-ai/gonzalo#15", "shared"), Some("board-b"));
        assert_ne!(ka, kb);
        assert!(store.get(&ka).await.unwrap().is_some());
        assert!(store.get(&kb).await.unwrap().is_some());
    }

    /// A store whose `put` always conflicts, to exercise the conflict arm.
    struct ConflictStore;

    #[async_trait::async_trait]
    impl gonzalo_core::Store for ConflictStore {
        async fn get(
            &self,
            _key: &gonzalo_core::RecordKey,
        ) -> gonzalo_core::Result<Option<Record>> {
            Ok(None)
        }

        async fn put(
            &self,
            record: Record,
            expected: Option<Revision>,
        ) -> gonzalo_core::Result<PutResult> {
            Ok(PutResult::Conflict(Box::new(gonzalo_core::Conflict {
                key: record.key.clone(),
                expected,
                current: record,
            })))
        }

        async fn list(
            &self,
            _prefix: &gonzalo_core::KeyPrefix,
        ) -> gonzalo_core::Result<Vec<gonzalo_core::RecordKey>> {
            Ok(vec![])
        }

        async fn delete_as(
            &self,
            _key: &gonzalo_core::RecordKey,
            _expected: Option<Revision>,
            _author: Option<Identity>,
        ) -> gonzalo_core::Result<gonzalo_core::DeleteResult> {
            Ok(gonzalo_core::DeleteResult::Deleted)
        }

        async fn put_raw(
            &self,
            record: Record,
            expected: Option<Revision>,
        ) -> gonzalo_core::Result<PutResult> {
            <Self as gonzalo_core::Store>::put(self, record, expected).await
        }

        async fn get_raw(
            &self,
            _key: &gonzalo_core::RecordKey,
        ) -> gonzalo_core::Result<Option<Record>> {
            Ok(None)
        }

        async fn list_raw(
            &self,
            _prefix: &gonzalo_core::KeyPrefix,
        ) -> gonzalo_core::Result<Vec<gonzalo_core::RecordKey>> {
            Ok(vec![])
        }

        async fn purge(
            &self,
            _key: &gonzalo_core::RecordKey,
            _expected: Revision,
        ) -> gonzalo_core::Result<gonzalo_core::DeleteResult> {
            Ok(gonzalo_core::DeleteResult::Deleted)
        }
    }

    #[tokio::test]
    async fn surfaces_store_conflict() {
        let src = InMemorySource::new(vec![ticket("a", "A")]);
        let err = ingest(&src, &ConflictStore, "tester", None)
            .await
            .unwrap_err();
        assert!(matches!(err, IngestError::Conflict(_)));
    }
}