use chrono::{DateTime, Utc};
use cliban_core::time;
use rusqlite::{params, Connection, Row};
use cliban_core::{Error, Result};
pub use cliban_core::migrations::REMOTE_LINKS_DDL as DDL;
const COLS: &str = "id, provider, entity, local_id, remote_id, remote_key, \
remote_updated_at, base_hash, last_synced_at, origin, \
progress_comment_id";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Origin {
#[default]
Imported,
Pushed,
}
impl Origin {
pub fn as_str(self) -> &'static str {
match self {
Origin::Imported => "imported",
Origin::Pushed => "pushed",
}
}
pub fn from_db(s: &str) -> Origin {
match s {
"pushed" => Origin::Pushed,
_ => Origin::Imported,
}
}
}
#[derive(Debug, Clone)]
pub struct RemoteLink {
pub id: i64,
pub provider: String,
pub entity: String,
pub local_id: i64,
pub remote_id: String,
pub remote_key: String,
pub remote_updated_at: Option<DateTime<Utc>>,
pub base_hash: Option<String>,
pub last_synced_at: DateTime<Utc>,
pub origin: Origin,
pub progress_comment_id: Option<String>,
}
#[derive(Debug, Clone)]
pub struct NewLink {
pub provider: String,
pub entity: String,
pub local_id: i64,
pub remote_id: String,
pub remote_key: String,
pub remote_updated_at: Option<DateTime<Utc>>,
pub base_hash: Option<String>,
pub origin: Origin,
}
pub fn ensure_table(conn: &Connection) -> Result<()> {
for stmt in DDL {
conn.execute_batch(stmt)?;
}
if !has_column(conn, "remote_links", "origin")? {
conn.execute_batch(
"ALTER TABLE remote_links ADD COLUMN origin TEXT NOT NULL DEFAULT 'imported'",
)?;
}
if !has_column(conn, "remote_links", "progress_comment_id")? {
conn.execute_batch("ALTER TABLE remote_links ADD COLUMN progress_comment_id TEXT")?;
}
Ok(())
}
fn has_column(conn: &Connection, table: &str, column: &str) -> Result<bool> {
let mut stmt = conn.prepare("SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2")?;
Ok(stmt.exists(params![table, column])?)
}
pub fn by_local(
conn: &Connection,
provider: &str,
entity: &str,
local_id: i64,
) -> Result<Option<RemoteLink>> {
let sql = format!(
"SELECT {COLS} FROM remote_links \
WHERE provider = ?1 AND entity = ?2 AND local_id = ?3"
);
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(params![provider, entity, local_id])?;
match rows.next()? {
Some(row) => Ok(Some(read(row)?)),
None => Ok(None),
}
}
pub fn by_remote(
conn: &Connection,
provider: &str,
entity: &str,
remote_id: &str,
) -> Result<Option<RemoteLink>> {
let sql = format!(
"SELECT {COLS} FROM remote_links \
WHERE provider = ?1 AND entity = ?2 AND remote_id = ?3"
);
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(params![provider, entity, remote_id])?;
match rows.next()? {
Some(row) => Ok(Some(read(row)?)),
None => Ok(None),
}
}
pub fn upsert(conn: &Connection, new: NewLink) -> Result<RemoteLink> {
let now = time::format_usec(time::now_usec());
let remote_updated = new.remote_updated_at.map(time::format_usec);
conn.execute(
"INSERT INTO remote_links \
(provider, entity, local_id, remote_id, remote_key, \
remote_updated_at, base_hash, last_synced_at, inserted_at, updated_at, origin) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8, ?8, ?9) \
ON CONFLICT (provider, entity, local_id) DO UPDATE SET \
remote_id = excluded.remote_id, \
remote_key = excluded.remote_key, \
remote_updated_at = excluded.remote_updated_at, \
base_hash = excluded.base_hash, \
last_synced_at = excluded.last_synced_at, \
updated_at = excluded.updated_at",
params![
new.provider,
new.entity,
new.local_id,
new.remote_id,
new.remote_key,
remote_updated,
new.base_hash,
now,
new.origin.as_str(),
],
)?;
by_local(conn, &new.provider, &new.entity, new.local_id)?.ok_or(Error::NotFound)
}
pub fn set_progress_comment(
conn: &Connection,
provider: &str,
entity: &str,
local_id: i64,
comment_id: Option<&str>,
) -> Result<()> {
let n = conn.execute(
"UPDATE remote_links SET progress_comment_id = ?4, updated_at = ?5 \
WHERE provider = ?1 AND entity = ?2 AND local_id = ?3",
params![
provider,
entity,
local_id,
comment_id,
time::format_usec(time::now_usec()),
],
)?;
if n == 0 {
return Err(Error::NotFound);
}
Ok(())
}
pub fn list(conn: &Connection, provider: &str) -> Result<Vec<RemoteLink>> {
let sql = format!("SELECT {COLS} FROM remote_links WHERE provider = ?1 ORDER BY id");
let mut stmt = conn.prepare(&sql)?;
let mut out = Vec::new();
let mut rows = stmt.query(params![provider])?;
while let Some(row) = rows.next()? {
out.push(read(row)?);
}
Ok(out)
}
fn read(row: &Row) -> Result<RemoteLink> {
let remote_updated_at: Option<String> = row.get(6)?;
let last_synced_at: String = row.get(8)?;
let origin: String = row.get(9)?;
Ok(RemoteLink {
id: row.get(0)?,
provider: row.get(1)?,
entity: row.get(2)?,
local_id: row.get(3)?,
remote_id: row.get(4)?,
remote_key: row.get(5)?,
remote_updated_at: remote_updated_at.as_deref().and_then(time::parse_ts),
base_hash: row.get(7)?,
last_synced_at: time::parse_ts(&last_synced_at).unwrap_or_else(time::now_usec),
origin: Origin::from_db(&origin),
progress_comment_id: row.get(10)?,
})
}
#[cfg(test)]
mod tests {
use super::*;
fn conn() -> Connection {
let c = Connection::open_in_memory().unwrap();
ensure_table(&c).unwrap();
c
}
fn new_link(local_id: i64, remote_id: &str, remote_key: &str) -> NewLink {
NewLink {
provider: crate::PROVIDER_LINEAR.into(),
entity: crate::ENTITY_ISSUE.into(),
local_id,
remote_id: remote_id.into(),
remote_key: remote_key.into(),
remote_updated_at: Some(time::now_usec()),
base_hash: Some("abc123".into()),
origin: Origin::Imported,
}
}
#[test]
fn ensure_table_is_idempotent() {
let c = conn();
ensure_table(&c).unwrap();
ensure_table(&c).unwrap();
}
#[test]
fn upsert_then_read_back_by_either_side() {
let c = conn();
let saved = upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
assert_eq!(saved.local_id, 7);
assert_eq!(saved.remote_key, "ENG-412");
let by_l = by_local(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 7)
.unwrap()
.unwrap();
let by_r = by_remote(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, "uuid-1")
.unwrap()
.unwrap();
assert_eq!(by_l.id, by_r.id);
assert_eq!(by_l.base_hash.as_deref(), Some("abc123"));
}
#[test]
fn upsert_refreshes_rather_than_duplicating() {
let c = conn();
upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
let again = upsert(&c, new_link(7, "uuid-1", "PLAT-9")).unwrap();
assert_eq!(again.remote_key, "PLAT-9");
assert_eq!(list(&c, crate::PROVIDER_LINEAR).unwrap().len(), 1);
}
#[test]
fn two_local_issues_cannot_claim_one_remote_issue() {
let c = conn();
upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
assert!(upsert(&c, new_link(8, "uuid-1", "ENG-412")).is_err());
}
#[test]
fn missing_link_reads_as_none_not_error() {
let c = conn();
assert!(
by_local(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 99)
.unwrap()
.is_none()
);
}
#[test]
fn ensure_table_adds_origin_to_a_pre_origin_table_and_backfills_imported() {
let c = Connection::open_in_memory().unwrap();
c.execute_batch(
r#"CREATE TABLE "remote_links" (
"id" INTEGER PRIMARY KEY AUTOINCREMENT,
"provider" TEXT NOT NULL,
"entity" TEXT NOT NULL,
"local_id" INTEGER NOT NULL,
"remote_id" TEXT NOT NULL,
"remote_key" TEXT NOT NULL,
"remote_updated_at" TEXT,
"base_hash" TEXT,
"last_synced_at" TEXT NOT NULL,
"inserted_at" TEXT NOT NULL,
"updated_at" TEXT NOT NULL
)"#,
)
.unwrap();
c.execute(
"INSERT INTO remote_links \
(provider, entity, local_id, remote_id, remote_key, last_synced_at, \
inserted_at, updated_at) \
VALUES ('linear', 'issue', 7, 'uuid-1', 'ENG-412', '2026-01-01T00:00:00Z', \
'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')",
[],
)
.unwrap();
ensure_table(&c).unwrap();
ensure_table(&c).unwrap();
let link = by_local(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 7)
.unwrap()
.unwrap();
assert_eq!(
link.origin,
Origin::Imported,
"pre-origin rows were only ever created by import"
);
}
#[test]
fn upsert_records_origin_and_a_later_sync_never_flips_it() {
let c = conn();
let mut first = new_link(7, "uuid-1", "ENG-412");
first.origin = Origin::Pushed;
let saved = upsert(&c, first).unwrap();
assert_eq!(saved.origin, Origin::Pushed);
let again = upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
assert_eq!(again.origin, Origin::Pushed);
}
#[test]
fn ensure_table_adds_progress_comment_id_to_a_pre_column_table() {
let c = Connection::open_in_memory().unwrap();
c.execute_batch(
r#"CREATE TABLE "remote_links" (
"id" INTEGER PRIMARY KEY AUTOINCREMENT,
"provider" TEXT NOT NULL,
"entity" TEXT NOT NULL,
"local_id" INTEGER NOT NULL,
"remote_id" TEXT NOT NULL,
"remote_key" TEXT NOT NULL,
"remote_updated_at" TEXT,
"base_hash" TEXT,
"last_synced_at" TEXT NOT NULL,
"inserted_at" TEXT NOT NULL,
"updated_at" TEXT NOT NULL,
"origin" TEXT NOT NULL DEFAULT 'imported'
)"#,
)
.unwrap();
c.execute(
"INSERT INTO remote_links \
(provider, entity, local_id, remote_id, remote_key, last_synced_at, \
inserted_at, updated_at) \
VALUES ('linear', 'issue', 7, 'uuid-1', 'ENG-412', '2026-01-01T00:00:00Z', \
'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')",
[],
)
.unwrap();
ensure_table(&c).unwrap();
ensure_table(&c).unwrap();
let link = by_local(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 7)
.unwrap()
.unwrap();
assert_eq!(
link.progress_comment_id, None,
"a pre-column row has no living comment yet"
);
}
#[test]
fn set_progress_comment_persists_and_reads_back() {
let c = conn();
upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
set_progress_comment(
&c,
crate::PROVIDER_LINEAR,
crate::ENTITY_ISSUE,
7,
Some("comment-uuid-1"),
)
.unwrap();
let link = by_local(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 7)
.unwrap()
.unwrap();
assert_eq!(link.progress_comment_id.as_deref(), Some("comment-uuid-1"));
set_progress_comment(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 7, None).unwrap();
let link = by_local(&c, crate::PROVIDER_LINEAR, crate::ENTITY_ISSUE, 7)
.unwrap()
.unwrap();
assert_eq!(link.progress_comment_id, None);
}
#[test]
fn upsert_never_clobbers_a_stored_comment_id() {
let c = conn();
upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
set_progress_comment(
&c,
crate::PROVIDER_LINEAR,
crate::ENTITY_ISSUE,
7,
Some("comment-uuid-1"),
)
.unwrap();
let again = upsert(&c, new_link(7, "uuid-1", "ENG-412")).unwrap();
assert_eq!(again.progress_comment_id.as_deref(), Some("comment-uuid-1"));
}
#[test]
fn an_unrecognized_origin_reads_as_imported() {
assert_eq!(Origin::from_db("imported"), Origin::Imported);
assert_eq!(Origin::from_db("pushed"), Origin::Pushed);
assert_eq!(Origin::from_db("mystery"), Origin::Imported);
}
}