use rusqlite::{params, Connection, OptionalExtension};
use tracing::debug;
use crate::core::errors::{Result, TgaError};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TicketTransitionRow {
pub ticket_key: String,
pub project_key: String,
pub from_status: Option<String>,
pub to_status: String,
pub transitioned_at: String,
pub author: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommentDetailRow {
pub ticket_key: String,
pub comment_id: String,
pub project_key: String,
pub author: Option<String>,
pub created_at: String,
pub body_len: i64,
}
pub fn upsert_ticket_transition(conn: &Connection, row: &TicketTransitionRow) -> Result<()> {
let synced_at = chrono::Utc::now().timestamp();
conn.execute(
"INSERT OR REPLACE INTO fact_ticket_transitions \
(ticket_key, project_key, from_status, to_status, transitioned_at, author, synced_at) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
row.ticket_key,
row.project_key,
row.from_status,
row.to_status,
row.transitioned_at,
row.author,
synced_at,
],
)
.map_err(TgaError::from)?;
debug!(
ticket = %row.ticket_key,
to_status = %row.to_status,
transitioned_at = %row.transitioned_at,
"upserted ticket transition"
);
Ok(())
}
pub fn upsert_comment_detail(conn: &Connection, row: &CommentDetailRow) -> Result<()> {
let synced_at = chrono::Utc::now().timestamp();
conn.execute(
"INSERT OR REPLACE INTO fact_jira_comment_detail \
(ticket_key, comment_id, project_key, author, created_at, body_len, synced_at) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
row.ticket_key,
row.comment_id,
row.project_key,
row.author,
row.created_at,
row.body_len,
synced_at,
],
)
.map_err(TgaError::from)?;
debug!(
ticket = %row.ticket_key,
comment_id = %row.comment_id,
"upserted comment detail"
);
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JiraSyncCursor {
pub last_synced_at: String,
pub last_run_at: String,
pub tickets_synced: i64,
}
pub fn get_cursor(conn: &Connection, project_key: &str) -> Result<Option<JiraSyncCursor>> {
conn.query_row(
"SELECT last_synced_at, last_run_at, tickets_synced \
FROM jira_sync_cursor WHERE project_key = ?1",
params![project_key],
|row| {
Ok(JiraSyncCursor {
last_synced_at: row.get(0)?,
last_run_at: row.get(1)?,
tickets_synced: row.get(2)?,
})
},
)
.optional()
.map_err(TgaError::from)
}
pub fn set_cursor(
conn: &Connection,
project_key: &str,
last_synced_at: &str,
tickets_synced: i64,
) -> Result<()> {
let now = chrono::Utc::now().to_rfc3339();
conn.execute(
"INSERT OR REPLACE INTO jira_sync_cursor \
(project_key, last_synced_at, last_run_at, tickets_synced) \
VALUES (?1, ?2, ?3, ?4)",
params![project_key, last_synced_at, now, tickets_synced],
)
.map_err(TgaError::from)?;
debug!(project = %project_key, last_synced_at, tickets_synced, "recorded jira sync cursor");
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FreshnessStatus {
pub table: &'static str,
pub project: Option<String>,
pub row_count: i64,
pub max_synced_at: Option<i64>,
pub age_seconds: Option<i64>,
pub stale: bool,
}
pub fn list_cursor_projects(conn: &Connection) -> Result<Vec<String>> {
let mut stmt = conn
.prepare("SELECT project_key FROM jira_sync_cursor ORDER BY project_key")
.map_err(TgaError::from)?;
let rows = stmt
.query_map([], |row| row.get::<_, String>(0))
.map_err(TgaError::from)?;
let mut out = Vec::new();
for row in rows {
out.push(row.map_err(TgaError::from)?);
}
Ok(out)
}
pub fn check_freshness(
conn: &Connection,
max_age_days: i64,
project_key: Option<&str>,
) -> Result<Vec<FreshnessStatus>> {
let now = chrono::Utc::now().timestamp();
let threshold_seconds = max_age_days.saturating_mul(86_400);
let mut out = Vec::with_capacity(2);
for table in ["fact_ticket_transitions", "fact_jira_comment_detail"] {
let sql = match project_key {
Some(_) => {
format!("SELECT COUNT(*), MAX(synced_at) FROM {table} WHERE project_key = ?1")
}
None => format!("SELECT COUNT(*), MAX(synced_at) FROM {table}"),
};
let map_row = |row: &rusqlite::Row<'_>| Ok((row.get(0)?, row.get(1)?));
let (row_count, max_synced_at): (i64, Option<i64>) = match project_key {
Some(key) => conn.query_row(&sql, params![key], map_row),
None => conn.query_row(&sql, [], map_row),
}
.map_err(TgaError::from)?;
let age_seconds = max_synced_at.map(|t| now - t);
let stale = match age_seconds {
None => true, Some(age) => age > threshold_seconds,
};
out.push(FreshnessStatus {
table,
project: project_key.map(str::to_string),
row_count,
max_synced_at,
age_seconds,
stale,
});
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::db::Database;
fn sample_transition(key: &str, to_status: &str, at: &str) -> TicketTransitionRow {
TicketTransitionRow {
ticket_key: key.to_string(),
project_key: "PROJ".to_string(),
from_status: Some("To Do".to_string()),
to_status: to_status.to_string(),
transitioned_at: at.to_string(),
author: Some("Jane Doe".to_string()),
}
}
fn sample_comment(key: &str, id: &str) -> CommentDetailRow {
CommentDetailRow {
ticket_key: key.to_string(),
comment_id: id.to_string(),
project_key: "PROJ".to_string(),
author: Some("Jane Doe".to_string()),
created_at: "2026-01-01T00:00:00Z".to_string(),
body_len: 42,
}
}
#[test]
fn upsert_transition_then_query_roundtrips() {
let db = Database::open_in_memory().expect("open");
let row = sample_transition("PROJ-1", "In Progress", "2026-01-01T00:00:00Z");
upsert_ticket_transition(db.connection(), &row).expect("upsert");
let (count, to_status): (i64, String) = db
.connection()
.query_row(
"SELECT COUNT(*), to_status FROM fact_ticket_transitions WHERE ticket_key = ?1",
params!["PROJ-1"],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.expect("query");
assert_eq!(count, 1);
assert_eq!(to_status, "In Progress");
}
#[test]
fn upsert_transition_is_idempotent_on_grain_key() {
let db = Database::open_in_memory().expect("open");
let row = sample_transition("PROJ-1", "In Progress", "2026-01-01T00:00:00Z");
upsert_ticket_transition(db.connection(), &row).expect("first");
upsert_ticket_transition(db.connection(), &row).expect("second");
let count: i64 = db
.connection()
.query_row(
"SELECT COUNT(*) FROM fact_ticket_transitions WHERE ticket_key = ?1",
params!["PROJ-1"],
|r| r.get(0),
)
.expect("count");
assert_eq!(
count, 1,
"re-inserting the same grain key must not duplicate"
);
}
#[test]
fn upsert_transition_keeps_repeat_status_visits_distinct() {
let db = Database::open_in_memory().expect("open");
upsert_ticket_transition(
db.connection(),
&sample_transition("PROJ-2", "Done", "2026-01-01T00:00:00Z"),
)
.expect("first done");
upsert_ticket_transition(
db.connection(),
&sample_transition("PROJ-2", "Done", "2026-02-01T00:00:00Z"),
)
.expect("reopened then re-done");
let count: i64 = db
.connection()
.query_row(
"SELECT COUNT(*) FROM fact_ticket_transitions WHERE ticket_key = 'PROJ-2'",
[],
|r| r.get(0),
)
.expect("count");
assert_eq!(
count, 2,
"revisiting the same to_status at a new time is a distinct row"
);
}
#[test]
fn upsert_comment_then_query_roundtrips() {
let db = Database::open_in_memory().expect("open");
let row = sample_comment("PROJ-1", "1001");
upsert_comment_detail(db.connection(), &row).expect("upsert");
let (count, body_len): (i64, i64) = db
.connection()
.query_row(
"SELECT COUNT(*), body_len FROM fact_jira_comment_detail WHERE ticket_key = ?1",
params!["PROJ-1"],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.expect("query");
assert_eq!(count, 1);
assert_eq!(body_len, 42);
}
#[test]
fn upsert_comment_is_idempotent_on_grain_key() {
let db = Database::open_in_memory().expect("open");
let row = sample_comment("PROJ-1", "1001");
upsert_comment_detail(db.connection(), &row).expect("first");
let mut updated = row.clone();
updated.body_len = 100;
upsert_comment_detail(db.connection(), &updated).expect("second");
let count: i64 = db
.connection()
.query_row(
"SELECT COUNT(*) FROM fact_jira_comment_detail WHERE ticket_key = 'PROJ-1'",
[],
|r| r.get(0),
)
.expect("count");
assert_eq!(
count, 1,
"re-upserting the same comment id must not duplicate"
);
let body_len: i64 = db
.connection()
.query_row(
"SELECT body_len FROM fact_jira_comment_detail WHERE ticket_key = 'PROJ-1'",
[],
|r| r.get(0),
)
.expect("read back");
assert_eq!(body_len, 100, "re-upsert must refresh the row's fields");
}
#[test]
fn cursor_round_trips_and_is_absent_initially() {
let db = Database::open_in_memory().expect("open");
assert!(get_cursor(db.connection(), "PROJ")
.expect("query")
.is_none());
set_cursor(db.connection(), "PROJ", "2026-01-01T00:00:00Z", 42).expect("set");
let cursor = get_cursor(db.connection(), "PROJ")
.expect("query")
.expect("present");
assert_eq!(cursor.last_synced_at, "2026-01-01T00:00:00Z");
assert_eq!(cursor.tickets_synced, 42);
set_cursor(db.connection(), "PROJ", "2026-02-01T00:00:00Z", 7).expect("overwrite");
let cursor2 = get_cursor(db.connection(), "PROJ")
.expect("query")
.expect("present");
assert_eq!(cursor2.last_synced_at, "2026-02-01T00:00:00Z");
assert_eq!(cursor2.tickets_synced, 7);
}
#[test]
fn cursor_is_scoped_per_project() {
let db = Database::open_in_memory().expect("open");
set_cursor(db.connection(), "PROJ", "2026-01-01T00:00:00Z", 1).expect("set proj");
set_cursor(db.connection(), "OTHER", "2026-03-01T00:00:00Z", 2).expect("set other");
assert_eq!(
get_cursor(db.connection(), "PROJ")
.expect("query")
.unwrap()
.last_synced_at,
"2026-01-01T00:00:00Z"
);
assert_eq!(
get_cursor(db.connection(), "OTHER")
.expect("query")
.unwrap()
.last_synced_at,
"2026-03-01T00:00:00Z"
);
}
#[test]
fn freshness_reports_stale_when_tables_are_empty() {
let db = Database::open_in_memory().expect("open");
let statuses = check_freshness(db.connection(), 2, None).expect("check");
assert_eq!(statuses.len(), 2);
for s in &statuses {
assert_eq!(s.row_count, 0);
assert!(s.max_synced_at.is_none());
assert!(s.stale, "{} must be stale when empty", s.table);
}
}
#[test]
fn freshness_reports_fresh_for_recently_synced_row() {
let db = Database::open_in_memory().expect("open");
upsert_ticket_transition(
db.connection(),
&sample_transition("PROJ-1", "Done", "2026-01-01T00:00:00Z"),
)
.expect("upsert transition");
upsert_comment_detail(db.connection(), &sample_comment("PROJ-1", "1"))
.expect("upsert comment");
let statuses = check_freshness(db.connection(), 2, None).expect("check");
for s in &statuses {
assert_eq!(s.row_count, 1);
assert!(s.max_synced_at.is_some());
assert!(!s.stale, "{} must be fresh right after a write", s.table);
}
}
#[test]
fn freshness_reports_stale_when_synced_at_is_older_than_threshold() {
let db = Database::open_in_memory().expect("open");
upsert_ticket_transition(
db.connection(),
&sample_transition("PROJ-1", "Done", "2026-01-01T00:00:00Z"),
)
.expect("upsert");
let ten_days_ago = chrono::Utc::now().timestamp() - 10 * 86_400;
db.connection()
.execute(
"UPDATE fact_ticket_transitions SET synced_at = ?1 WHERE ticket_key = 'PROJ-1'",
params![ten_days_ago],
)
.expect("backdate");
let statuses = check_freshness(db.connection(), 2, None).expect("check");
let transitions = statuses
.iter()
.find(|s| s.table == "fact_ticket_transitions")
.expect("present");
assert!(
transitions.stale,
"a row synced 10 days ago must be stale against a 2-day threshold"
);
assert!(transitions.age_seconds.unwrap() >= 10 * 86_400);
}
#[test]
fn freshness_is_project_scoped() {
let db = Database::open_in_memory().expect("open");
let mut healthy = sample_transition("A-1", "Done", "2026-01-01T00:00:00Z");
healthy.project_key = "A".into();
upsert_ticket_transition(db.connection(), &healthy).expect("upsert A");
let mut healthy_comment = sample_comment("A-1", "1");
healthy_comment.project_key = "A".into();
upsert_comment_detail(db.connection(), &healthy_comment).expect("upsert A comment");
let mut dead = sample_transition("B-1", "Done", "2026-01-01T00:00:00Z");
dead.project_key = "B".into();
upsert_ticket_transition(db.connection(), &dead).expect("upsert B");
let ten_days_ago = chrono::Utc::now().timestamp() - 10 * 86_400;
db.connection()
.execute(
"UPDATE fact_ticket_transitions SET synced_at = ?1 WHERE project_key = 'B'",
params![ten_days_ago],
)
.expect("backdate B");
let unscoped = check_freshness(db.connection(), 2, None).expect("unscoped");
let unscoped_transitions = unscoped
.iter()
.find(|s| s.table == "fact_ticket_transitions")
.expect("present");
assert!(
!unscoped_transitions.stale,
"the table-wide aggregate is exactly the blind spot: A's write hides B"
);
let scoped = check_freshness(db.connection(), 2, Some("B")).expect("scoped");
let scoped_transitions = scoped
.iter()
.find(|s| s.table == "fact_ticket_transitions")
.expect("present");
assert!(
scoped_transitions.stale,
"project B's dead sync must surface once the check is scoped to it"
);
assert_eq!(scoped_transitions.project.as_deref(), Some("B"));
}
#[test]
fn list_cursor_projects_returns_every_synced_project() {
let db = Database::open_in_memory().expect("open");
assert!(list_cursor_projects(db.connection())
.expect("query")
.is_empty());
set_cursor(db.connection(), "B", "2026-01-01T00:00:00Z", 1).expect("set B");
set_cursor(db.connection(), "A", "2026-01-01T00:00:00Z", 1).expect("set A");
assert_eq!(
list_cursor_projects(db.connection()).expect("query"),
vec!["A".to_string(), "B".to_string()]
);
}
}