Something went wrong. Try again.
unified development activity
Something went wrong. Try again.
148 kB · 3666 lines
Rust
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667//! SQLite storage.//!//! The important invariant lives in the schema, not in a cleanup job: a commit//! is one row keyed by its Git object name, and the places it was seen are rows//! in `commit_refs`. Tangled may report the same commit on a dozen branches and//! on every resubmit of a pull request; all of those collapse to one `commits`//! row. That is what makes the headline number "actual commits" rather than//! "things an indexer said".
use std::collections::{HashMap, HashSet};use std::path::Path;use std::sync::{Arc, Mutex};
use anyhow::{Context, Result};use chrono::{DateTime, Utc};use rusqlite::{Connection, OptionalExtension, params};
use crate::model::{Branch, Commit, CommitRef, Identity, IdentityKind, Repo};use crate::reach::{RepoState, RunCoverage, Standing};use crate::shard::Shard;
/// Each repository's own rows are found through `repos`' primary key:/// `p.repo_did = substr(r.repo_key, 9)` strips the `tangled:` prefix from the/// reference's key. The reverse form, `r.repo_key = 'tangled:' || p.repo_did`,/// can't use an index (the comparison takes `repo_key`'s text affinity, the/// expression has none), so SQLite scanned every repository for every/// reference: hours on a full global index instead of minutes.////// The per-day count the global summary used before counted changes were/// exported; kept as the reference the export is tested against.#[cfg(test)]const TANGLED_DAILY_COUNTS_SQL: &str = r#" select day, count(*) from ( select substr(min(c.authored_at), 1, 10) as day from commits c where exists ( select 1 from counted_refs r where r.sha = c.sha and r.source = 'tangled' and not exists ( select 1 from repos p where p.excluded = 1 and p.repo_did = substr(r.repo_key, 9) ) and not exists ( select 1 from repos p where p.created_at is not null and p.repo_did = substr(r.repo_key, 9) and date(c.authored_at) < date(p.created_at, ?1) ) ) group by case when c.change_id != '' then c.change_id else c.sha end ) group by day order by day"#;
/// The six columns of a counted-changes export: the earliest UTC author day a/// logical change counts on, for everyone, the owner's identities and their/// agents, counting the default branch only and then every branch.pub const COUNTED_COLUMNS: [&str; 6] = [ "all_default", "owner_default", "agent_default", "all_any", "owner_any", "agent_any",];
/// Every counted Tangled logical change per repository, keyed by Change-Id,/// else (with `?3` = 1) by author email, author time and subject for a/// rewritten commit, else by SHA, in one pass over the references. A/// reference doesn't count in a repository a manual rule excludes or whose/// verdict isn't to count it (a held one is kept aside, see/// [`write_counted_rows`]), nor for a commit authored before its repository/// existed (less the grace period `?1`, none for a repo that lost it) or/// after `?2`. A day is the earliest author date's first ten characters, and/// the minimum of prefixes is the prefix of the minimum, so this matches/// grouping first. `{shard}` narrows the references to one shard's/// repositories.const REPO_CHANGES_SQL: &str = r#" select substr(r.repo_key, 9) as repo, case when c.change_id != '' then c.change_id when ?3 = 1 and c.author_email != '' then 'rebased:' || lower(c.author_email) || char(31) || c.authored_at || char(31) || c.message else c.sha end as k, coalesce(e.verdict, 'include'), min(case when r.on_default = 1 then substr(c.authored_at, 1, 10) end), min(case when r.on_default = 1 and c.author_did in (select did from temp.counted_owner) then substr(c.authored_at, 1, 10) end), min(case when r.on_default = 1 and c.author_did in (select did from temp.counted_agent) then substr(c.authored_at, 1, 10) end), min(substr(c.authored_at, 1, 10)), min(case when c.author_did in (select did from temp.counted_owner) then substr(c.authored_at, 1, 10) end), min(case when c.author_did in (select did from temp.counted_agent) then substr(c.authored_at, 1, 10) end) from commit_refs r join commits c on c.sha = r.sha left join repos p on p.repo_did = substr(r.repo_key, 9) left join repo_evidence e on e.repo_did = substr(r.repo_key, 9) where r.source = 'tangled' and coalesce(p.excluded, 0) = 0 and coalesce(e.verdict, 'include') in ('include', 'empty', 'hold') and coalesce(date(c.authored_at) < date(p.created_at, case when e.no_grace = 1 then '-0 days' else ?1 end), 0) = 0 and c.authored_at <= ?2 {shard} group by r.repo_key, k"#;
const REPO_LOGICAL_COUNTS_SQL: &str = " select r.repo_key, count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commits c join counted_refs r on r.sha = c.sha where r.source = 'tangled' and not exists ( select 1 from repos p where p.excluded = 1 and p.repo_did = substr(r.repo_key, 9) ) group by r.repo_key";
/// Personal GitHub identities from before reports stopped naming the login.const LEGACY_PERSONAL_GITHUB: &str = "kind = 'github' and did like 'github:%' and did not in ('github:personal', 'github:work')";
/// Store a reference; a later sighting on the default branch upgrades an/// earlier one from a side branch, never the other way round.const INSERT_REF: &str = "insert into commit_refs(sha, source, repo_key, url, on_default) values (?1, ?2, ?3, ?4, ?5) on conflict(sha, source, repo_key) do update set on_default = max(on_default, excluded.on_default)";
/// Store a reference from a report that is exact about branches: its marker/// replaces the stored one, so a commit an older report counted as merged/// stops counting once a newer one says it is only on a side branch.const REPLACE_REF: &str = "insert into commit_refs(sha, source, repo_key, url, on_default) values (?1, ?2, ?3, ?4, ?5) on conflict(sha, source, repo_key) do update set on_default = excluded.on_default";
/// Which branches' commits count. Every query that counts reads/// `counted_refs`, a view this selects; storage keeps every branch either/// way, so switching needs no recrawl.#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum BranchScope { /// Only commits reachable from a repository's default branch: merged /// work. Rebased or abandoned side branches don't count. Default, /// Commits reachable from any branch. All,}
impl BranchScope { pub fn parse(value: &str) -> Result<Self> { match value { "default" => Ok(Self::Default), "all" => Ok(Self::All), other => anyhow::bail!("TALLY_BRANCH_SCOPE must be default or all, not {other:?}"), } }
pub fn as_str(self) -> &'static str { match self { Self::Default => "default", Self::All => "all", } }}
fn create_counted_refs(conn: &Connection, scope: BranchScope) -> Result<()> { let filter = match scope { BranchScope::Default => "where on_default = 1", BranchScope::All => "", }; conn.execute_batch(&format!( "drop view if exists temp.counted_refs; create temp view counted_refs as select * from commit_refs {filter};" ))?; Ok(())}
#[derive(Debug, Clone)]pub struct Db { conn: Arc<Mutex<Connection>>, scope: Arc<Mutex<BranchScope>>,}
impl Db { pub fn open(path: impl AsRef<Path>) -> Result<Self> { let conn = Connection::open(path.as_ref()) .with_context(|| format!("opening {}", path.as_ref().display()))?; let db = Self { conn: Arc::new(Mutex::new(conn)), scope: Arc::new(Mutex::new(BranchScope::Default)), }; db.migrate()?; db.set_branch_scope(BranchScope::Default)?; Ok(db) }
/// Fold the write-ahead log into the database file, so the file alone is /// the whole database. Uploading it is then safe for as long as nothing /// writes: the crawl's checkpoint holds the only connection while it /// uploads, so no copy of a multi-gigabyte file is needed first. pub fn flush_wal(&self) -> Result<()> { let conn = self.conn.lock().unwrap(); let (busy, _, _): (i64, i64, i64) = conn.query_row("pragma wal_checkpoint(truncate)", [], |row| { Ok((row.get(0)?, row.get(1)?, row.get(2)?)) })?; if busy != 0 { anyhow::bail!("the write-ahead log could not be checkpointed"); } Ok(()) }
/// Create the database at `path` for one shard of the global crawl from /// the single-task database at `seed`: every repository's metadata, and /// only this shard's branches, references and commits. A shard then /// resumes where the single task left off instead of reading Tangled /// from the start, from a file a fraction of the size. pub fn seed_shard(path: &Path, seed: &Path, shard: Shard) -> Result<Self> { // Bring the seed to this schema first, so the copy can name columns. drop(Self::open(seed).with_context(|| format!("opening seed {}", seed.display()))?); let db = Self::open(path)?; { let conn = db.conn.lock().unwrap(); // One large transaction into a new file: a rollback journal has // nothing to save, where the write-ahead log would hold a second // copy of every page until the next checkpoint. conn.pragma_update(None, "journal_mode", "DELETE")?; conn.execute( "attach database ?1 as seed", [seed.to_string_lossy().as_ref()], )?; let copied = seed_shard_rows(&conn, shard); let detached = conn.execute("detach database seed", []); copied?; detached?; conn.pragma_update(None, "journal_mode", "WAL")?; } Ok(db) }
#[cfg(test)] pub fn in_memory() -> Result<Self> { let db = Self { conn: Arc::new(Mutex::new(Connection::open_in_memory()?)), scope: Arc::new(Mutex::new(BranchScope::Default)), }; db.migrate()?; db.set_branch_scope(BranchScope::Default)?; Ok(db) }
fn migrate(&self) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.pragma_update(None, "journal_mode", "WAL")?; conn.pragma_update(None, "synchronous", "NORMAL")?; conn.busy_timeout(std::time::Duration::from_secs(10))?; conn.execute_batch( r#" create table if not exists meta ( key text primary key, value text not null );
create table if not exists identities ( did text primary key, handle text not null default '', kind text not null, operator_did text, color text not null default '', active integer not null default 1, pds text, collections text not null default '', checked_at text, last_seen text not null );
create table if not exists repos ( repo_did text primary key, owner_did text not null default '', name text not null default '', knot text not null default '', url text not null default '', website text, fork integer not null default 0, created_at text, excluded integer not null default 0, exclude_reason text, active integer not null default 1, first_seen text not null, last_seen text not null, last_scanned text, last_error text ); create index if not exists repos_last_scanned on repos(last_scanned);
create table if not exists branches ( repo_did text not null, name text not null, head_sha text not null default '', indexed integer not null default 0, last_scanned text, primary key (repo_did, name) );
create table if not exists commits ( sha text primary key, change_id text not null default '', author_did text not null default '', author_email text not null default '', author_name text not null default '', authored_at text not null default '', committed_at text not null default '', message text not null default '', first_source text not null default '' ); create index if not exists commits_author on commits(author_did); create index if not exists commits_authored on commits(authored_at);
create table if not exists commit_refs ( sha text not null, source text not null, repo_key text not null, url text not null default '', primary key (sha, source, repo_key) ); create index if not exists commit_refs_source on commit_refs(source); create index if not exists commit_refs_repo on commit_refs(repo_key);
-- Each repository's evidence and verdict under the global rule -- set (see `rules`), replaced as a whole on every run. create table if not exists repo_evidence ( repo_did text primary key, verdict text not null, reason text not null, no_grace integer not null default 0, commits integer not null default 0, authors integer not null default 0, tangled_commits integer not null default 0, pre_creation integer not null default 0, pre_creation_authors integer not null default 0, history_before integer not null default 0, signals text not null default '', rules_hash text not null );
create table if not exists github_repos ( full_name text primary key, pushed_at text not null default '', homepage text, last_scanned text, last_error text );
-- One row per vibescrobble record that carried usage. create table if not exists scrobble_usage ( did text not null, rkey text not null, created_at text not null, model text not null, input integer not null, output integer not null, cache_read integer not null, cache_write integer not null, primary key (did, rkey) ) without rowid;
create table if not exists scrobble_polls ( did text primary key, polled_at text not null ); "#, )?;
// Older databases predate the logical-change column. `create table if // not exists` does not add it, so migrate explicitly. let has_change_id = { let mut stmt = conn.prepare("pragma table_info(commits)")?; let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; let mut found = false; for name in rows { if name? == "change_id" { found = true; break; } } found }; if !has_change_id { conn.execute( "alter table commits add column change_id text not null default ''", [], )?; }
let has_repo_website = { let mut stmt = conn.prepare("pragma table_info(repos)")?; let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; let mut found = false; for name in rows { if name? == "website" { found = true; break; } } found }; if !has_repo_website { conn.execute("alter table repos add column website text", [])?; }
let has_on_default = { let mut stmt = conn.prepare("pragma table_info(commit_refs)")?; let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; let mut found = false; for name in rows { if name? == "on_default" { found = true; break; } } found }; if !has_on_default { conn.execute( "alter table commit_refs add column on_default integer not null default 0", [], )?; // Nothing stored so far says which branch it came from, so every // branch is walked once more: the default branch marks its history // as it goes, and side branches stop at once in known history. conn.execute("update branches set indexed = 0", [])?; // GitHub references counted on every branch until now, and keep // counting until a newer scan or report marks them. conn.execute( "update commit_refs set on_default = 1 where source != 'tangled'", [], )?; }
let has_walk_head = { let mut stmt = conn.prepare("pragma table_info(branches)")?; let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; let mut found = false; for name in rows { if name? == "walk_head" { found = true; break; } } found }; if !has_walk_head { // Where an unfinished walk is. Walks from before this have no // saved place; they start again from their head. conn.execute_batch( "alter table branches add column walk_head text; alter table branches add column walk_offset integer not null default 0;", )?; }
let has_github_homepage = { let mut stmt = conn.prepare("pragma table_info(github_repos)")?; let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; let mut found = false; for name in rows { if name? == "homepage" { found = true; break; } } found }; if !has_github_homepage { conn.execute("alter table github_repos add column homepage text", [])?; }
for (name, ddl) in [ ( "fork", "alter table repos add column fork integer not null default 0", ), ("created_at", "alter table repos add column created_at text"), ( "excluded", "alter table repos add column excluded integer not null default 0", ), ( "exclude_reason", "alter table repos add column exclude_reason text", ), ( "description", "alter table repos add column description text", ), ( "history_before", "alter table repos add column history_before integer not null default 0", ), // How the crawl found the repository (search, a legacy search // record, an owner's own records) and its run of failures, for // the global summary's coverage. ( "discovered_via", "alter table repos add column discovered_via text", ), ( "last_success", "alter table repos add column last_success text; update repos set last_success = last_scanned where last_error is null and last_scanned is not null", ), ( "fail_count", "alter table repos add column fail_count integer not null default 0; alter table repos add column failing_since text; update repos set fail_count = 1, failing_since = last_scanned where last_error is not null", ), ] { let mut stmt = conn.prepare("pragma table_info(repos)")?; let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; let mut found = false; for column in rows { if column? == name { found = true; break; } } if !found { conn.execute_batch(ddl)?; } }
// Personal GitHub reports were once stored under `github:<login>`. // They are `github:personal` now, but commits keep the author they // were first stored with, so the old identity kept them and the site // listed "personal GitHub" twice. Fold the old one in. conn.execute_batch(&format!( "insert or ignore into identities (did, handle, kind, color, active, collections, last_seen) select 'github:personal', 'personal GitHub', 'github', color, 1, '', last_seen from identities where {LEGACY_PERSONAL_GITHUB} limit 1; update commits set author_did = 'github:personal' where author_did in (select did from identities where {LEGACY_PERSONAL_GITHUB}); delete from identities where {LEGACY_PERSONAL_GITHUB};" ))?; Ok(()) }
pub fn get_meta(&self, key: &str) -> Result<Option<String>> { let conn = self.conn.lock().unwrap(); Ok(conn .query_row("select value from meta where key = ?1", [key], |row| { row.get(0) }) .optional()?) }
pub fn set_meta(&self, key: &str, value: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "insert into meta(key, value) values (?1, ?2) on conflict(key) do update set value = excluded.value", params![key, value], )?; Ok(()) }
pub fn upsert_identity(&self, identity: &Identity) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( r#" insert into identities (did, handle, kind, operator_did, color, active, pds, collections, checked_at, last_seen) values (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) on conflict(did) do update set handle = excluded.handle, kind = excluded.kind, operator_did = coalesce(excluded.operator_did, identities.operator_did), color = excluded.color, active = excluded.active, pds = coalesce(excluded.pds, identities.pds), collections = case when excluded.collections = '' then identities.collections else excluded.collections end, checked_at = coalesce(excluded.checked_at, identities.checked_at), last_seen = excluded.last_seen "#, params![ identity.did, identity.handle, identity.kind.as_str(), identity.operator_did, identity.color, identity.active as i64, identity.pds, identity.collections, identity.checked_at, identity.checked_at.clone().unwrap_or_else(now_string), ], )?; Ok(()) }
pub fn mark_identity_inactive(&self, did: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "update identities set active = 0 where did = ?1", params![did], )?; Ok(()) }
pub fn list_identities(&self) -> Result<Vec<Identity>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select did, handle, kind, operator_did, color, active, pds, collections, checked_at from identities order by kind, handle", )?; let rows = stmt.query_map([], |row| { Ok(Identity { did: row.get(0)?, handle: row.get(1)?, kind: IdentityKind::parse(&row.get::<_, String>(2)?), operator_did: row.get(3)?, color: row.get(4)?, active: row.get::<_, i64>(5)? != 0, pds: row.get(6)?, collections: row.get(7)?, checked_at: row.get(8)?, }) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
pub fn get_identity(&self, did: &str) -> Result<Option<Identity>> { Ok(self .list_identities()? .into_iter() .find(|identity| identity.did == did)) }
pub fn upsert_repo(&self, repo: &Repo) -> Result<()> { let now = now_string(); let conn = self.conn.lock().unwrap(); conn.execute( r#" insert into repos (repo_did, owner_did, name, knot, url, website, fork, created_at, active, first_seen, last_seen) values (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) on conflict(repo_did) do update set owner_did = case when excluded.owner_did = '' then repos.owner_did else excluded.owner_did end, name = case when excluded.name = '' then repos.name else excluded.name end, knot = case when excluded.knot = '' then repos.knot else excluded.knot end, url = case when excluded.url = '' then repos.url else excluded.url end, website = coalesce(excluded.website, repos.website), fork = excluded.fork, created_at = coalesce(excluded.created_at, repos.created_at), active = excluded.active, last_seen = excluded.last_seen "#, params![ repo.repo_did, repo.owner_did, repo.name, repo.knot, repo.url, repo.website, repo.fork as i64, repo.created_at, repo.active as i64, now, now, ], )?; Ok(()) }
/// Record an attempt to read `repo_did`: a success clears its run of /// failures, a failure extends it (see `reach::classify`). pub fn mark_repo_scanned(&self, repo_did: &str, error: Option<&str>) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "update repos set last_scanned = ?2, last_error = ?3, last_success = case when ?3 is null then ?2 else last_success end, fail_count = case when ?3 is null then 0 else fail_count + 1 end, failing_since = case when ?3 is null then null else coalesce(failing_since, ?2) end where repo_did = ?1", params![repo_did, now_string(), error], )?; Ok(()) }
/// Note how this run found `repo_did`. pub fn set_discovered_via(&self, repo_did: &str, via: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "update repos set discovered_via = ?2 where repo_did = ?1", params![repo_did, via], )?; Ok(()) }
/// Mark one repository excluded, as a manual rule would. Runs replace /// every exclusion through [`Db::store_evidence`]; tests set one directly. #[cfg(test)] pub fn exclude_repo(&self, repo_did: &str, reason: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "update repos set excluded = 1, exclude_reason = ?2, last_error = null where repo_did = ?1", params![repo_did, reason], )?; Ok(()) }
/// Store a repository's description from its record, for the /// described-copy rule. An absent description leaves the stored one. pub fn set_repo_description(&self, repo_did: &str, description: Option<&str>) -> Result<()> { let Some(description) = description else { return Ok(()); }; let conn = self.conn.lock().unwrap(); conn.execute( "update repos set description = ?2 where repo_did = ?1", params![repo_did, description], )?; Ok(()) }
/// Remember that a walk passed `seen` commits from before the repo's /// floor: its history began before it was on Tangled. Keeps the most any /// walk saw, since an incremental walk stops long before that history. pub fn note_history_before(&self, repo_did: &str, seen: u64) -> Result<()> { if seen == 0 { return Ok(()); } let conn = self.conn.lock().unwrap(); conn.execute( "update repos set history_before = max(history_before, ?2) where repo_did = ?1", params![repo_did, seen as i64], )?; Ok(()) }
/// What the rules know about every repository: its record and its own /// commits from its grace floor (`imported_before`, and its creation day /// with `grace`, a date modifier) up to `cap`, every branch. pub fn repo_facts( &self, imported_before: &str, grace: &str, cap: &str, ) -> Result<Vec<crate::rules::RepoFacts>> { use crate::rules::RepoFacts; let conn = self.conn.lock().unwrap(); let mut facts: HashMap<String, RepoFacts> = HashMap::new(); let mut owners: HashMap<String, u64> = HashMap::new(); let mut stmt = conn.prepare( "select repo_did, owner_did, fork, description, coalesce(last_error, '') like 'truncated after %', history_before from repos", )?; let mut rows = stmt.query([])?; while let Some(row) = rows.next()? { let repo = RepoFacts { repo_did: row.get(0)?, owner_did: row.get(1)?, fork: row.get::<_, i64>(2)? != 0, description: row.get(3)?, truncated: row.get(4)?, history_before: row.get::<_, i64>(5)? as u64, ..Default::default() }; if !repo.fork && !repo.owner_did.is_empty() { *owners.entry(repo.owner_did.clone()).or_default() += 1; } facts.insert(repo.repo_did.clone(), repo); } drop(rows); drop(stmt); let mut stmt = conn.prepare( "select substr(r.repo_key, 9), count(*), count(distinct lower(c.author_email)), sum(c.author_email like 'did:%' or c.author_did != ''), sum(coalesce(date(c.authored_at) < date(p.created_at), 0)), count(distinct case when date(c.authored_at) < date(p.created_at) then lower(c.author_email) end) from commit_refs r join commits c on c.sha = r.sha left join repos p on p.repo_did = substr(r.repo_key, 9) where r.source = 'tangled' and c.authored_at >= ?1 and c.authored_at <= ?3 and coalesce(date(c.authored_at) < date(p.created_at, ?2), 0) = 0 group by r.repo_key", )?; let mut rows = stmt.query(params![imported_before, grace, cap])?; while let Some(row) = rows.next()? { let did: String = row.get(0)?; let repo = facts.entry(did.clone()).or_insert_with(|| RepoFacts { repo_did: did, ..Default::default() }); repo.commits = row.get::<_, i64>(1)? as u64; repo.authors = row.get::<_, i64>(2)? as u64; repo.tangled_commits = row.get::<_, i64>(3)? as u64; repo.pre_creation = row.get::<_, i64>(4)? as u64; repo.pre_creation_authors = row.get::<_, i64>(5)? as u64; } drop(rows); drop(stmt); // History from before the floor that an earlier, unfiltered walk // stored counts as seen too. let mut stmt = conn.prepare( "select substr(r.repo_key, 9), count(*) from commit_refs r join commits c on c.sha = r.sha left join repos p on p.repo_did = substr(r.repo_key, 9) where r.source = 'tangled' and (c.authored_at < ?1 or coalesce(date(c.authored_at) < date(p.created_at, ?2), 0) = 1) group by r.repo_key", )?; let mut rows = stmt.query(params![imported_before, grace])?; while let Some(row) = rows.next()? { let did: String = row.get(0)?; if let Some(repo) = facts.get_mut(&did) { repo.history_before = repo.history_before.max(row.get::<_, i64>(1)? as u64); } } drop(rows); drop(stmt); let mut out: Vec<RepoFacts> = facts .into_values() .map(|mut repo| { repo.owner_repos = owners.get(&repo.owner_did).copied().unwrap_or(0); repo }) .collect(); out.sort_by(|a, b| a.repo_did.cmp(&b.repo_did)); Ok(out) }
/// Replace every repository's verdict with `decisions`, all in one /// transaction: the exclusions a manual rule makes (the crawl skips those /// repos) and the evidence table the count reads. A repo no rule /// excludes any more is excluded no more. pub fn store_evidence( &self, decisions: &[(crate::rules::RepoFacts, crate::rules::Evidence)], rules_hash: &str, ) -> Result<()> { use crate::rules::Signal; let mut conn = self.conn.lock().unwrap(); let tx = conn.transaction()?; tx.execute( "update repos set excluded = 0, exclude_reason = null where excluded != 0", [], )?; tx.execute("delete from repo_evidence", [])?; { let mut exclude = tx.prepare( "update repos set excluded = 1, exclude_reason = ?2 where repo_did = ?1", )?; let mut insert = tx.prepare( "insert into repo_evidence (repo_did, verdict, reason, no_grace, commits, authors, tangled_commits, pre_creation, pre_creation_authors, history_before, signals, rules_hash) values (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", )?; for (facts, evidence) in decisions { // The reasons the crawl has always written for these. let manual = if evidence.has(Signal::ExcludesFile) { Some("excludes file") } else if evidence.has(Signal::ForkRecord) { Some("fork") } else if evidence.has(Signal::DescribedCopy) { facts .description .as_deref() .and_then(crate::described::copy_reason) } else { None }; if let Some(reason) = manual { exclude.execute(params![facts.repo_did, reason])?; } let signals = evidence .signals .iter() .map(|signal| signal.as_str()) .collect::<Vec<_>>() .join(","); insert.execute(params![ facts.repo_did, evidence.verdict.as_str(), evidence.reason, evidence.no_grace as i64, facts.commits as i64, facts.authors as i64, facts.tangled_commits as i64, facts.pre_creation as i64, facts.pre_creation_authors as i64, facts.history_before as i64, signals, rules_hash, ])?; } } tx.commit()?; Ok(()) }
pub fn list_active_repos(&self) -> Result<Vec<Repo>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select repo_did, owner_did, name, knot, url, website, fork, created_at, active, last_scanned, last_error from repos where active = 1 and excluded = 0", )?; let rows = stmt.query_map([], |row| { Ok(Repo { repo_did: row.get(0)?, owner_did: row.get(1)?, name: row.get(2)?, knot: row.get(3)?, url: row.get(4)?, website: row.get(5)?, fork: row.get::<_, i64>(6)? != 0, created_at: row.get(7)?, active: row.get::<_, i64>(8)? != 0, last_scanned: row.get(9)?, last_error: row.get(10)?, }) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
pub fn get_branch(&self, repo_did: &str, name: &str) -> Result<Option<Branch>> { let conn = self.conn.lock().unwrap(); Ok(conn .query_row( "select repo_did, name, head_sha, indexed, walk_head, walk_offset from branches where repo_did = ?1 and name = ?2", params![repo_did, name], |row| { Ok(Branch { repo_did: row.get(0)?, name: row.get(1)?, head_sha: row.get(2)?, indexed: row.get::<_, i64>(3)? != 0, walk_head: row.get(4)?, walk_offset: row.get::<_, i64>(5)?.max(0) as u64, }) }, ) .optional()?) }
pub fn upsert_branch( &self, repo_did: &str, name: &str, head_sha: &str, indexed: bool, ) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( r#" insert into branches(repo_did, name, head_sha, indexed, last_scanned) values (?1, ?2, ?3, ?4, ?5) on conflict(repo_did, name) do update set head_sha = excluded.head_sha, indexed = excluded.indexed, last_scanned = excluded.last_scanned, walk_head = case when excluded.indexed = 1 then null else walk_head end, walk_offset = case when excluded.indexed = 1 then 0 else walk_offset end "#, params![repo_did, name, head_sha, indexed as i64, now_string()], )?; Ok(()) }
/// Save where a walk of `name` is: `offset` commits into the log of /// `head`. The branch stays unindexed until the walk completes. pub fn set_walk_cursor( &self, repo_did: &str, name: &str, head: &str, offset: u64, ) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "insert into branches (repo_did, name, head_sha, indexed, walk_head, walk_offset) values (?1, ?2, ?3, 0, ?3, ?4) on conflict(repo_did, name) do update set indexed = 0, walk_head = ?3, walk_offset = ?4", params![repo_did, name, head, offset as i64], )?; Ok(()) }
/// The branches of `repo_did` with a walk an earlier run didn't finish. pub fn unfinished_walks(&self, repo_did: &str) -> Result<HashSet<String>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn .prepare("select name from branches where repo_did = ?1 and walk_head is not null")?; let rows = stmt.query_map([repo_did], |row| row.get::<_, String>(0))?; Ok(rows.collect::<rusqlite::Result<HashSet<_>>>()?) }
/// Insert a commit and the place it was seen. Returns true when the commit /// row itself was new. /// /// A commit that already exists but had no resolved identity is upgraded in /// place: identity resolution can improve after the first sighting (a new /// bot account, a newly configured email), and the commit is the same /// commit, not a new one. #[cfg(test)] pub fn insert_commit(&self, commit: &Commit, reference: &CommitRef) -> Result<bool> { let conn = self.conn.lock().unwrap(); let inserted = conn.execute( r#" insert or ignore into commits (sha, change_id, author_did, author_email, author_name, authored_at, committed_at, message, first_source) values (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) "#, params![ commit.sha, commit.change_id, commit.author_did, commit.author_email, commit.author_name, commit.authored_at, commit.committed_at, commit.message, reference.source, ], )? > 0;
if !inserted && !commit.author_did.is_empty() { conn.execute( "update commits set author_did = ?2, author_email = ?3, author_name = ?4 where sha = ?1 and author_did = ''", params![ commit.sha, commit.author_did, commit.author_email, commit.author_name ], )?; } if !inserted && !commit.change_id.is_empty() { conn.execute( "update commits set change_id = ?2 where sha = ?1 and change_id = ''", params![commit.sha, commit.change_id], )?; }
conn.execute( INSERT_REF, params![ commit.sha, reference.source, reference.repo_key, reference.url, reference.on_default ], )?; Ok(inserted) }
/// Batch insert in one transaction. Returns the number of newly seen SHAs. pub fn insert_commits(&self, rows: &[(Commit, CommitRef)]) -> Result<usize> { self.insert_rows(rows, INSERT_REF) }
/// Like [`Db::insert_commits`], for a report whose default-branch /// markers are exact and replace what earlier reports said. pub fn insert_report_commits(&self, rows: &[(Commit, CommitRef)]) -> Result<usize> { self.insert_rows(rows, REPLACE_REF) }
/// True when a stored reference under `repo_key` was seen only on a side /// branch, so a merge could still make it count. pub fn has_side_branch_refs(&self, source: &str, repo_key: &str) -> Result<bool> { let conn = self.conn.lock().unwrap(); Ok(conn.query_row( "select exists(select 1 from commit_refs where source = ?1 and repo_key = ?2 and on_default = 0)", params![source, repo_key], |row| row.get(0), )?) }
fn insert_rows(&self, rows: &[(Commit, CommitRef)], ref_sql: &str) -> Result<usize> { let mut conn = self.conn.lock().unwrap(); let tx = conn.transaction()?; let mut new = 0usize; for (commit, reference) in rows { let inserted = tx.execute( r#" insert or ignore into commits (sha, change_id, author_did, author_email, author_name, authored_at, committed_at, message, first_source) values (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) "#, params![ commit.sha, commit.change_id, commit.author_did, commit.author_email, commit.author_name, commit.authored_at, commit.committed_at, commit.message, reference.source, ], )?; if inserted > 0 { new += 1; } else if !commit.author_did.is_empty() { tx.execute( "update commits set author_did = ?2, author_email = ?3, author_name = ?4 where sha = ?1 and author_did = ''", params![ commit.sha, commit.author_did, commit.author_email, commit.author_name ], )?; } if !commit.change_id.is_empty() { tx.execute( "update commits set change_id = ?2 where sha = ?1 and change_id = ''", params![commit.sha, commit.change_id], )?; } tx.execute( ref_sql, params![ commit.sha, reference.source, reference.repo_key, reference.url, reference.on_default ], )?; } tx.commit()?; Ok(new) }
/// Fill in authorship for commits that were seen before the identity was /// known. `emails` maps lower-case address to DID; `handles` maps Tangled /// handle (without `@`) to DID. pub fn reclassify_commits( &self, emails: &HashMap<String, String>, handles: &HashMap<String, String>, ) -> Result<usize> { // Avoid holding the connection lock while building SQL; the sets are // small (four people, a handful of aliases). let conn = self.conn.lock().unwrap(); let mut changed = 0usize; if !emails.is_empty() { let mut stmt = conn.prepare( "select sha, author_email, author_name from commits where author_did = ''", )?; let rows = stmt.query_map([], |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, )) })?; let mut updates = Vec::new(); for row in rows { let (sha, email, name) = row?; let email_key = email.to_ascii_lowercase(); let did = emails.get(&email_key).cloned().or_else(|| { name.strip_prefix('@') .and_then(|handle| handles.get(handle)) .cloned() }); if let Some(did) = did { updates.push((sha, did)); } } for (sha, did) in updates { changed += conn.execute( "update commits set author_did = ?2 where sha = ?1 and author_did = ''", params![sha, did], )?; } } Ok(changed) }
#[cfg(test)] pub fn tangled_total(&self) -> Result<u64> { let conn = self.conn.lock().unwrap(); Ok(conn.query_row( "select count(distinct r.sha) from counted_refs r where r.source = 'tangled'", [], |row| row.get::<_, i64>(0), )? as u64) }
#[cfg(test)] pub fn github_total(&self) -> Result<u64> { let conn = self.conn.lock().unwrap(); Ok(conn.query_row( "select count(distinct r.sha) from counted_refs r where r.source = 'github'", [], |row| row.get::<_, i64>(0), )? as u64) }
#[cfg(test)] pub fn commits_for_dids(&self, dids: &[String], source: &str) -> Result<u64> { if dids.is_empty() { return Ok(0); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select count(distinct c.sha) from commits c join counted_refs r on r.sha = c.sha where r.source = ?1 and c.author_did in ({placeholders})" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); values.push(Box::new(source.to_string())); for did in dids { values.push(Box::new(did.clone())); } let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); Ok(conn.query_row(&sql, refs.as_slice(), |row| row.get::<_, i64>(0))? as u64) }
/// Count distinct commits matching an optional author set, source and /// lower bound. An empty DID slice means everyone. pub fn count_commits( &self, dids: &[String], source: Option<&str>, since: Option<&str>, ) -> Result<u64> { let conn = self.conn.lock().unwrap(); let mut sql = String::from( "select count(distinct c.sha) from commits c join counted_refs r on r.sha = c.sha where 1 = 1", ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(source) = source { sql.push_str(" and r.source = ?"); values.push(Box::new(source.to_string())); } if let Some(since) = since { sql.push_str(" and c.authored_at >= ?"); values.push(Box::new(since.to_string())); } if !dids.is_empty() { let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); sql.push_str(&format!(" and c.author_did in ({placeholders})")); for did in dids { values.push(Box::new(did.clone())); } }
let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); Ok(conn.query_row(&sql, refs.as_slice(), |row| row.get::<_, i64>(0))? as u64) }
/// Every commit already indexed for one repository. Used as the stop /// boundary when walking a second branch through shared history. pub fn repo_commit_shas(&self, repo_did: &str) -> Result<HashSet<String>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare("select sha from commit_refs where repo_key = ?1 and source = 'tangled'")?; let rows = stmt.query_map([format!("tangled:{repo_did}")], |row| { row.get::<_, String>(0) })?; Ok(rows.collect::<rusqlite::Result<HashSet<_>>>()?) }
/// The repository's commits already marked as on its default branch: the /// stop boundary for the default branch's own walk. pub fn repo_default_shas(&self, repo_did: &str) -> Result<HashSet<String>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select sha from commit_refs where repo_key = ?1 and source = 'tangled' and on_default = 1", )?; let rows = stmt.query_map([format!("tangled:{repo_did}")], |row| { row.get::<_, String>(0) })?; Ok(rows.collect::<rusqlite::Result<HashSet<_>>>()?) }
/// Mark commits stored from another branch as on this repository's /// default branch, once its walk reaches them. pub fn mark_on_default(&self, repo_did: &str, shas: &[String]) -> Result<()> { if shas.is_empty() { return Ok(()); } let mut conn = self.conn.lock().unwrap(); let tx = conn.transaction()?; let key = format!("tangled:{repo_did}"); for sha in shas { tx.execute( "update commit_refs set on_default = 1 where sha = ?1 and source = 'tangled' and repo_key = ?2", params![sha, key], )?; } tx.commit()?; Ok(()) }
/// Choose which branches' commits every count reads (see `BranchScope`). pub fn set_branch_scope(&self, scope: BranchScope) -> Result<()> { create_counted_refs(&self.conn.lock().unwrap(), scope)?; *self.scope.lock().unwrap() = scope; Ok(()) }
#[cfg(test)] pub fn conn_for_tests(&self) -> std::sync::MutexGuard<'_, Connection> { self.conn.lock().unwrap() }
#[cfg(test)] pub fn branch_scope(&self) -> BranchScope { *self.scope.lock().unwrap() }
/// Count logical changes: one row per `Change-Id` when present, otherwise /// per SHA. A rebase changes the SHA but keeps the Change-Id, so the /// rewritten object does not inflate this count. The date is the earliest /// author date seen for the logical change. pub fn count_logical_commits( &self, dids: &[String], source: Option<&str>, since: Option<&str>, ) -> Result<u64> { let conn = self.conn.lock().unwrap(); let mut sql = String::from( "select count(*) from ( select case when c.change_id != '' then c.change_id else c.sha end as k, min(c.authored_at) as first_at from commits c join counted_refs r on r.sha = c.sha where 1 = 1", ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); if let Some(source) = source { sql.push_str(" and r.source = ?"); values.push(Box::new(source.to_string())); } if !dids.is_empty() { let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); sql.push_str(&format!(" and c.author_did in ({placeholders})")); for did in dids { values.push(Box::new(did.clone())); } } sql.push_str(" group by k)"); if let Some(since) = since { sql.push_str(" where first_at >= ?"); values.push(Box::new(since.to_string())); } let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); Ok(conn.query_row(&sql, refs.as_slice(), |row| row.get::<_, i64>(0))? as u64) }
#[cfg(test)] /// Distinct logical Tangled changes per UTC author day by `dids`, under the /// same exclusions and repo-age filter as `tangled_daily_counts`, so the /// global summary can carry the owner's side of the share from the same /// snapshot as the denominator. Repositories are found by primary key /// (`p.repo_did = substr(r.repo_key, 9)`); comparing against /// `'tangled:' || p.repo_did` instead can't use an index. pub fn tangled_daily_counts_by( &self, dids: &[String], repo_age_grace_days: i64, ) -> Result<Vec<(String, u64)>> { if dids.is_empty() { return Ok(Vec::new()); } let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select day, count(*) from ( select substr(min(c.authored_at), 1, 10) as day from commits c where c.author_did in ({placeholders}) and exists ( select 1 from counted_refs r where r.sha = c.sha and r.source = 'tangled' and not exists ( select 1 from repos p where p.excluded = 1 and p.repo_did = substr(r.repo_key, 9) ) and not exists ( select 1 from repos p where p.created_at is not null and p.repo_did = substr(r.repo_key, 9) and date(c.authored_at) < date(p.created_at, ?) ) ) group by case when c.change_id != '' then c.change_id else c.sha end ) group by day order by day" ); let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare(&sql)?; let modifier = format!("-{repo_age_grace_days} days"); let mut values: Vec<&dyn rusqlite::ToSql> = dids.iter().map(|did| did as &dyn rusqlite::ToSql).collect(); values.push(&modifier); let rows = stmt.query_map(values.as_slice(), |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
pub fn identity_counts_logical(&self, source: &str) -> Result<HashMap<String, u64>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select c.author_did, count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commits c join counted_refs r on r.sha = c.sha where r.source = ?1 and c.author_did != '' group by c.author_did", )?; let rows = stmt.query_map([source], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<HashMap<_, _>>>()?) }
/// Daily logical changes, placed at the earliest author date seen for each /// Change-Id. This is what the matrix should draw. Each change names one /// repository, so per-project counts add up to the day's total: the /// earliest commit's, preferring Tangled over a GitHub mirror. pub fn daily_rows_logical(&self, since: &str) -> Result<Vec<DailyRow>> { let conn = self.conn.lock().unwrap(); // SQLite takes the bare columns from the row min() picked. let mut stmt = conn.prepare( r#" select author_did, authored_at, (select r.repo_key from counted_refs r where r.sha = logical.sha order by r.source = 'tangled' desc, r.repo_key limit 1) from ( select c.sha, c.author_did, min(c.authored_at) as authored_at from commits c where exists (select 1 from counted_refs r where r.sha = c.sha) group by case when c.change_id != '' then c.change_id else c.sha end ) logical where authored_at >= ?1 "#, )?; let rows = stmt.query_map([since], |row| { Ok(DailyRow { author_did: row.get(0)?, authored_at: row.get(1)?, repo_key: row.get(2)?, }) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// Each logical change since `since` as `(author_did, authored_at, /// subject)`, taken from its earliest commit, like the daily rows. pub fn subject_rows_logical(&self, since: &str) -> Result<Vec<(String, String, String)>> { let conn = self.conn.lock().unwrap(); // SQLite takes the bare columns from the row min() picked. let mut stmt = conn.prepare( r#" select author_did, authored_at, message from ( select c.author_did, min(c.authored_at) as authored_at, c.message from commits c where exists (select 1 from counted_refs r where r.sha = c.sha) group by case when c.change_id != '' then c.change_id else c.sha end ) where authored_at >= ?1 "#, )?; let rows = stmt.query_map([since], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// The logical changes behind `daily_rows_logical`, grouped and dated /// the same way, with every counted reference of the commit each one is /// taken from: one row per reference. pub fn day_rows_logical(&self, since: &str) -> Result<Vec<crate::day_shards::DayRef>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( r#" select l.sha, l.author_did, l.authored_at, l.message, r.repo_key, r.url from ( select c.sha, c.author_did, min(c.authored_at) as authored_at, c.message from commits c where exists (select 1 from counted_refs r where r.sha = c.sha) group by case when c.change_id != '' then c.change_id else c.sha end ) l join counted_refs r on r.sha = l.sha where l.authored_at >= ?1 order by l.authored_at, l.sha, r.repo_key "#, )?; let rows = stmt.query_map([since], |row| { Ok(crate::day_shards::DayRef { sha: row.get(0)?, author_did: row.get(1)?, authored_at: row.get(2)?, message: row.get(3)?, repo_key: row.get(4)?, url: row.get(5)?, }) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
#[cfg(test)] /// Distinct logical Tangled changes per UTC author day. This is the /// denominator the global task publishes; the hourly task never /// derives it from its own owned-only database. pub fn tangled_daily_counts(&self, repo_age_grace_days: i64) -> Result<Vec<(String, u64)>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare(TANGLED_DAILY_COUNTS_SQL)?; let modifier = format!("-{repo_age_grace_days} days"); let rows = stmt.query_map([modifier], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// Each counted Tangled logical change authored since `since`, as /// `(author_did, earliest authored_at, latest committed_at, subject, /// repo_key)`. pub fn land_rows_logical(&self, since: &str) -> Result<Vec<crate::land_gap::LandRow>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( r#" select author_did, authored_at, committed_at, message, repo_key from ( select min(c.author_did) as author_did, min(c.authored_at) as authored_at, max(c.committed_at) as committed_at, min(c.message) as message, min(r.repo_key) as repo_key from commits c join counted_refs r on r.sha = c.sha and r.source = 'tangled' group by case when c.change_id != '' then c.change_id else c.sha end ) where authored_at >= ?1 "#, )?; let rows = stmt.query_map([since], |row| { Ok(( row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?, )) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// One `(repo_key, authored_at)` row per logical change per repository /// by `dids` since a date, Tangled and GitHub alike, counted the same /// way as the project ranking. pub fn project_day_rows_logical( &self, dids: &[String], since: &str, ) -> Result<Vec<(String, String)>> { if dids.is_empty() { return Ok(Vec::new()); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select r.repo_key, min(c.authored_at) from commits c join counted_refs r on r.sha = c.sha where r.source in ('tangled', 'github') and c.authored_at >= ? and c.author_did in ({placeholders}) group by r.repo_key, case when c.change_id != '' then c.change_id else c.sha end" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(since.to_string())]; for did in dids { values.push(Box::new(did.clone())); } let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map(refs.as_slice(), |row| Ok((row.get(0)?, row.get(1)?)))?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// Logical Tangled commit count per repository key. Used by the /// filter-only audit to find already-indexed oversized histories without /// touching the network. pub fn repo_logical_counts(&self) -> Result<Vec<(String, u64)>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare(REPO_LOGICAL_COUNTS_SQL)?; let rows = stmt.query_map([], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// Total commits and the subset authored by `dids` in one repo. pub fn repo_commit_counts(&self, repo_key: &str, dids: &[String]) -> Result<(u64, u64)> { let conn = self.conn.lock().unwrap(); let total: i64 = conn.query_row( "select count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commits c join counted_refs r on r.sha = c.sha where r.repo_key = ?1 and r.source = 'tangled'", [repo_key], |row| row.get(0), )?; let mine = if dids.is_empty() { 0 } else { let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commits c join counted_refs r on r.sha = c.sha where r.repo_key = ?1 and r.source = 'tangled' and c.author_did in ({placeholders})" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); values.push(Box::new(repo_key.to_string())); for did in dids { values.push(Box::new(did.clone())); } let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); conn.query_row(&sql, refs.as_slice(), |row| row.get::<_, i64>(0))? }; Ok((total as u64, mine as u64)) }
/// Per-identity all-time object counts. Used by the redaction test; the /// dashboard uses the logical variant. #[cfg(test)] pub fn identity_counts(&self, source: &str) -> Result<HashMap<String, u64>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select c.author_did, count(distinct c.sha) from commits c join counted_refs r on r.sha = c.sha where r.source = ?1 and c.author_did != '' group by c.author_did", )?; let rows = stmt.query_map([source], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<HashMap<_, _>>>()?) }
#[cfg(test)] pub fn top_repos(&self, limit: usize) -> Result<Vec<(String, u64)>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select repo_key, count(distinct sha) as n from counted_refs where source = 'tangled' group by repo_key order by n desc limit ?1", )?; let rows = stmt.query_map([limit as i64], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// Repositories ranked by commits authored by `dids` since a date. This is /// the project-ranking query; `top_repos_for_dids` is its all-time twin. pub fn top_repos_for_dids_since( &self, dids: &[String], since: &str, source: &str, limit: usize, ) -> Result<Vec<(String, u64)>> { if dids.is_empty() { return Ok(Vec::new()); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select r.repo_key, count(distinct case when c.change_id != '' then c.change_id else c.sha end) as n from commits c join counted_refs r on r.sha = c.sha where r.source = ? and c.authored_at >= ? and c.author_did in ({placeholders}) group by r.repo_key order by n desc limit ?" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); values.push(Box::new(source.to_string())); values.push(Box::new(since.to_string())); for did in dids { values.push(Box::new(did.clone())); } values.push(Box::new(limit as i64)); let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map(refs.as_slice(), |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// One repository's commits in a window. Works for Tangled and GitHub keys /// because the source is carried by the ref rows, not the key shape. pub fn repo_commit_counts_since( &self, repo_key: &str, dids: &[String], since: &str, ) -> Result<u64> { if dids.is_empty() { return Ok(0); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commits c join counted_refs r on r.sha = c.sha where r.repo_key = ? and r.source in ('tangled', 'github') and c.authored_at >= ? and c.author_did in ({placeholders})" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); values.push(Box::new(repo_key.to_string())); values.push(Box::new(since.to_string())); for did in dids { values.push(Box::new(did.clone())); } let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); Ok(conn.query_row(&sql, refs.as_slice(), |row| row.get::<_, i64>(0))? as u64) }
/// One repository's commits in a half-open window, used for period-over- /// period momentum without double-counting a Change-Id on the boundary. pub fn repo_commit_counts_between( &self, repo_key: &str, dids: &[String], from: &str, to: &str, ) -> Result<u64> { if dids.is_empty() { return Ok(0); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commits c join counted_refs r on r.sha = c.sha where r.repo_key = ? and r.source in ('tangled', 'github') and c.authored_at >= ? and c.authored_at < ? and c.author_did in ({placeholders})" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); values.push(Box::new(repo_key.to_string())); values.push(Box::new(from.to_string())); values.push(Box::new(to.to_string())); for did in dids { values.push(Box::new(did.clone())); } let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); Ok(conn.query_row(&sql, refs.as_slice(), |row| row.get::<_, i64>(0))? as u64) }
/// Repositories ranked by commits authored by `dids`, which is what a /// "where did my work go" panel needs. A global top-60 list would drop the /// person's smaller repositories entirely. pub fn top_repos_for_dids(&self, dids: &[String], limit: usize) -> Result<Vec<(String, u64)>> { if dids.is_empty() { return Ok(Vec::new()); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select r.repo_key, count(distinct case when c.change_id != '' then c.change_id else c.sha end) as n from commits c join counted_refs r on r.sha = c.sha where r.source = 'tangled' and c.author_did in ({placeholders}) group by r.repo_key order by n desc limit ?" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); for did in dids { values.push(Box::new(did.clone())); } values.push(Box::new(limit as i64)); let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map(refs.as_slice(), |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// GitHub repositories ranked by authored commits. Redacted work repos are /// returned under their opaque key and never joined back to a name. pub fn top_github_repos_for_dids( &self, dids: &[String], limit: usize, ) -> Result<Vec<(String, u64)>> { if dids.is_empty() { return Ok(Vec::new()); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( "select r.repo_key, count(distinct case when c.change_id != '' then c.change_id else c.sha end) as n from commits c join counted_refs r on r.sha = c.sha where r.source = 'github' and c.author_did in ({placeholders}) group by r.repo_key order by n desc limit ?" ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); for did in dids { values.push(Box::new(did.clone())); } values.push(Box::new(limit as i64)); let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map(refs.as_slice(), |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
/// Recent authored commits from either forge. Redacted repositories and /// the work inbox are deliberately absent: their rows have no message or /// URL to show, and the aggregate counts already include them. Named /// personal inbox repositories are included. pub fn recent_commits_for_dids_all( &self, dids: &[String], limit: usize, ) -> Result<Vec<RecentCommit>> { if dids.is_empty() { return Ok(Vec::new()); } let conn = self.conn.lock().unwrap(); let placeholders = (0..dids.len()).map(|_| "?").collect::<Vec<_>>().join(","); let sql = format!( r#" select c.sha, c.author_did, max(c.authored_at) as authored_at, c.message, r.repo_key, r.url from commits c join counted_refs r on r.sha = c.sha where r.source in ('tangled', 'github') and r.repo_key not like 'github:redacted:%' and r.repo_key not like 'github:inbox:work:%' and r.repo_key != 'github:inbox:work' and c.author_did in ({placeholders}) group by case when c.change_id != '' then c.change_id else c.sha end order by authored_at desc limit ? "# ); let mut values: Vec<Box<dyn rusqlite::ToSql>> = Vec::new(); for did in dids { values.push(Box::new(did.clone())); } values.push(Box::new(limit as i64)); let refs: Vec<&dyn rusqlite::ToSql> = values.iter().map(|value| value.as_ref()).collect(); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map(refs.as_slice(), |row| { Ok(RecentCommit { sha: row.get(0)?, author_did: row.get(1)?, authored_at: row.get(2)?, message: row.get(3)?, repo_key: row.get(4)?, url: row.get(5)?, }) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
pub fn repo_coverage(&self) -> Result<Coverage> { let conn = self.conn.lock().unwrap(); let total: i64 = conn.query_row( "select count(*) from repos where active = 1 and excluded = 0", [], |row| row.get(0), )?; let scanned: i64 = conn.query_row( "select count(*) from repos where active = 1 and excluded = 0 and last_scanned is not null and last_error is null", [], |row| row.get(0), )?; let failed: i64 = conn.query_row( "select count(*) from repos where active = 1 and excluded = 0 and last_error is not null", [], |row| row.get(0), )?; let branches: i64 = conn.query_row("select count(*) from branches", [], |row| row.get(0))?; let indexed_branches: i64 = conn.query_row( "select count(*) from branches where indexed = 1", [], |row| row.get(0), )?; Ok(Coverage { repos_total: total as u64, repos_scanned: scanned as u64, repos_failed: failed as u64, branches_total: branches as u64, branches_indexed: indexed_branches as u64, }) }
/// Repositories whose history no walk has finished, which are due however /// recently they were scanned: one with a branch still unindexed (the /// default-branch migration requeued every branch; a walk that failed, /// timed out or hit a deadline leaves one too), or with Tangled /// references none of which is marked as on its default branch. A /// repository stopped at the page cap is left to the refresh window: /// another walk would stop at the same cap. pub fn repos_needing_walk(&self) -> Result<HashSet<String>> { let mut needing = self.repos_with_unindexed_branches()?; let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select substr(repo_key, 9) from commit_refs where source = 'tangled' group by repo_key having max(on_default) = 0", )?; let rows = stmt.query_map([], |row| row.get::<_, String>(0))?; for row in rows { needing.insert(row?); } Ok(needing) }
/// Repositories with a branch no walk has finished, other than those /// stopped at the page cap. fn repos_with_unindexed_branches(&self) -> Result<HashSet<String>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select distinct b.repo_did from branches b left join repos p on p.repo_did = b.repo_did where b.indexed = 0 and coalesce(p.last_error, '') not like 'truncated after %'", )?; let rows = stmt.query_map([], |row| row.get::<_, String>(0))?; Ok(rows.collect::<rusqlite::Result<HashSet<_>>>()?) }
/// Forget the branches of `repo_did` that Tangled no longer lists, so a /// deleted branch that was never indexed doesn't keep the repository /// due forever. pub fn prune_branches(&self, repo_did: &str, current: &[&str]) -> Result<usize> { let conn = self.conn.lock().unwrap(); let stored: Vec<String> = conn .prepare("select name from branches where repo_did = ?1")? .query_map([repo_did], |row| row.get(0))? .collect::<rusqlite::Result<_>>()?; let mut removed = 0; for name in stored .iter() .filter(|name| !current.contains(&name.as_str())) { removed += conn.execute( "delete from branches where repo_did = ?1 and name = ?2", params![repo_did, name], )?; } Ok(removed) }
/// How many of one shard's repositories have a branch no walk has /// finished, for the publish guard. (A repository whose only commits are /// on side branches has nothing to mark, so it doesn't count here.) pub fn repos_unwalked_in(&self, shard: Shard) -> Result<u64> { let unwalked = self.repos_with_unindexed_branches()?; Ok(self .list_active_repos()? .iter() .filter(|repo| shard.contains(&repo.repo_did) && unwalked.contains(&repo.repo_did)) .count() as u64) }
/// This run's coverage of `shard`'s repositories (see [`RunCoverage`]): /// what was read since `scanned_since`, what failed, what can't be read, /// and an estimate of the recent changes the unread ones hold. pub fn run_coverage( &self, shard: Shard, scanned_since: &str, now: DateTime<Utc>, ) -> Result<RunCoverage> { struct Row { repo_did: String, knot: String, excluded: bool, last_scanned: Option<String>, last_error: Option<String>, fail_count: u32, failing_since: Option<String>, last_success: Option<String>, discovered_via: Option<String>, } let rows: Vec<Row> = { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select repo_did, knot, excluded, last_scanned, last_error, fail_count, failing_since, last_success, discovered_via from repos where active = 1", )?; stmt.query_map([], |row| { Ok(Row { repo_did: row.get(0)?, knot: row.get(1)?, excluded: row.get::<_, i64>(2)? != 0, last_scanned: row.get(3)?, last_error: row.get(4)?, fail_count: row.get::<_, i64>(5)?.max(0) as u32, failing_since: row.get(6)?, last_success: row.get(7)?, discovered_via: row.get(8)?, }) })? .collect::<rusqlite::Result<_>>()? }; let unwalked = self.repos_with_unindexed_branches()?; let mut coverage = RunCoverage { scanned_since: scanned_since.to_string(), ..RunCoverage::default() }; let mut unread = Vec::new(); for row in rows.iter().filter(|row| shard.contains(&row.repo_did)) { if row.excluded { coverage.repos_discovered += 1; coverage.repos_excluded += 1; continue; } *coverage .discovered_via .entry( row.discovered_via .as_deref() .unwrap_or("unknown") .to_string(), ) .or_default() += 1; let state = RepoState { knot: &row.knot, last_error: row.last_error.as_deref(), fail_count: row.fail_count, failing_since: row.failing_since.as_deref(), walk_unfinished: unwalked.contains(&row.repo_did), }; let standing = coverage.count( state, row.last_scanned.as_deref(), row.last_success.as_deref(), now, ); // A walk stopped at the page cap read the recent history; only // the oldest is missing, which no 30-day estimate covers. let truncated = row .last_error .as_deref() .is_some_and(crate::reach::is_truncated); if standing != Standing::Scanned && !truncated { unread.push((row, standing)); } } let (mut missing, mut unreachable) = (0.0, 0.0); for (row, standing) in unread { let key = format!("tangled:{}", row.repo_did); let last_good = match row .last_success .as_deref() .and_then(|at| DateTime::parse_from_rfc3339(at).ok()) { Some(at) => at.with_timezone(&Utc), // Never read successfully as far as this database knows: // its newest stored change stands in for the last read. None => match self.newest_change(&key, now)? { Some(at) => at, None => { if standing != Standing::Unreachable { coverage.missing_unknown_repos += 1; } continue; } }, }; let before = self.changes_between(&key, last_good - chrono::Duration::days(30), last_good)?; let estimate = crate::reach::missing_estimate(before, last_good, now); if standing == Standing::Unreachable { unreachable += estimate; } else { missing += estimate; } } coverage.missing_changes_30d_est = missing.round() as u64; coverage.unreachable_changes_30d_est = unreachable.round() as u64; Ok(coverage) }
/// Distinct logical changes of one repository authored in `[from, to)`, /// on any branch: a repository whose last good read predates /// default-branch marking has none marked, and an estimate that errs /// high beats one that reads zero. fn changes_between( &self, repo_key: &str, from: DateTime<Utc>, to: DateTime<Utc>, ) -> Result<u64> { let stamp = |at: DateTime<Utc>| at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let conn = self.conn.lock().unwrap(); let count: i64 = conn.query_row( "select count(distinct case when c.change_id != '' then c.change_id else c.sha end) from commit_refs r join commits c on c.sha = r.sha where r.repo_key = ?1 and c.authored_at >= ?2 and c.authored_at < ?3", params![repo_key, stamp(from), stamp(to)], |row| row.get(0), )?; Ok(count as u64) }
/// When one repository's newest stored change was authored, no later than /// `now` (a few commits carry future dates). fn newest_change(&self, repo_key: &str, now: DateTime<Utc>) -> Result<Option<DateTime<Utc>>> { let conn = self.conn.lock().unwrap(); let newest: Option<String> = conn.query_row( "select max(c.authored_at) from commit_refs r join commits c on c.sha = r.sha where r.repo_key = ?1", [repo_key], |row| row.get(0), )?; Ok(newest .as_deref() .and_then(|at| DateTime::parse_from_rfc3339(at).ok()) .map(|at| at.with_timezone(&Utc).min(now))) }
/// Write this database's counted Tangled changes to a new SQLite file at /// `out` (see [`COUNTED_COLUMNS`]), with `meta` beside them. `owner` and /// `agents` are the DIDs whose own columns are filled; `shard` keeps only /// references to its repositories; `rules` decide what counts, nothing /// authored after `cap` does, and `today` ends the windows a large /// repository is measured in. Returns the number of changes. #[allow(clippy::too_many_arguments)] pub fn write_counted( &self, out: &Path, owner: &[String], agents: &[String], rules: &crate::rules::RuleSet, cap: &str, today: chrono::NaiveDate, shard: Shard, meta: &[(&str, String)], ) -> Result<u64> { if out.exists() { std::fs::remove_file(out) .with_context(|| format!("replacing counted changes at {}", out.display()))?; } let conn = self.conn.lock().unwrap(); conn.execute( "attach database ?1 as counts", [out.to_string_lossy().as_ref()], )?; let written = write_counted_rows(&conn, owner, agents, rules, cap, today, shard, meta); let detached = conn.execute("detach database counts", []); let written = written?; detached?; Ok(written) }
pub fn error_summary(&self, limit: usize) -> Result<Vec<(String, String)>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select repo_did, last_error from repos where excluded = 0 and last_error is not null order by last_scanned desc limit ?1", )?; let rows = stmt.query_map([limit as i64], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }
pub fn repo_by_key(&self) -> Result<HashMap<String, Repo>> { Ok(self .list_active_repos()? .into_iter() .map(|repo| (format!("tangled:{}", repo.repo_did), repo)) .collect()) }
pub fn github_pushed_at(&self, full_name: &str) -> Result<Option<String>> { let conn = self.conn.lock().unwrap(); Ok(conn .query_row( "select pushed_at from github_repos where full_name = ?1", [full_name], |row| row.get(0), ) .optional()?) }
pub fn github_last_scanned(&self, full_name: &str) -> Result<Option<String>> { let conn = self.conn.lock().unwrap(); Ok(conn .query_row( "select last_scanned from github_repos where full_name = ?1", [full_name], |row| row.get(0), ) .optional()?) }
pub fn github_last_error(&self, full_name: &str) -> Result<Option<String>> { let conn = self.conn.lock().unwrap(); Ok(conn .query_row( "select last_error from github_repos where full_name = ?1", [full_name], |row| row.get(0), ) .optional()? .flatten()) }
pub fn touch_github_repo( &self, full_name: &str, pushed_at: &str, homepage: Option<&str>, error: Option<&str>, ) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( r#" insert into github_repos(full_name, pushed_at, homepage, last_scanned, last_error) values (?1, ?2, ?3, ?4, ?5) on conflict(full_name) do update set pushed_at = excluded.pushed_at, homepage = coalesce(excluded.homepage, github_repos.homepage), last_scanned = excluded.last_scanned, last_error = excluded.last_error "#, params![full_name, pushed_at, homepage, now_string(), error], )?; Ok(()) }
pub fn github_homepages(&self) -> Result<HashMap<String, String>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select full_name, homepage from github_repos where homepage is not null and homepage != ''", )?; let rows = stmt.query_map([], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) })?; Ok(rows.collect::<rusqlite::Result<HashMap<_, _>>>()?) }
pub fn scrobble_polls(&self) -> Result<HashMap<String, crate::burn::Poll>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select p.did, p.polled_at, max(u.created_at) from scrobble_polls p left join scrobble_usage u using (did) group by p.did", )?; let rows = stmt.query_map([], |row| Ok((row.get(0)?, (row.get(1)?, row.get(2)?))))?; Ok(rows.collect::<rusqlite::Result<HashMap<_, _>>>()?) }
/// Store one agent's records and when they were read. A record already /// stored keeps its row. pub fn store_scrobbles( &self, did: &str, rows: &[crate::burn::UsageRow], polled_at: &str, ) -> Result<()> { let mut conn = self.conn.lock().unwrap(); let tx = conn.transaction()?; for row in rows { let [input, output, cache_read, cache_write] = row.tokens; tx.execute( "insert or ignore into scrobble_usage values (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", params![ did, row.rkey, row.created_at, row.model, input, output, cache_read, cache_write ], )?; } tx.execute( "insert or replace into scrobble_polls values (?1, ?2)", params![did, polled_at], )?; tx.commit()?; Ok(()) }
pub fn prune_scrobbles(&self, before: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute("delete from scrobble_usage where created_at < ?1", [before])?; Ok(()) }
pub fn scrobble_usage(&self) -> Result<Vec<crate::burn::UsageRow>> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "select did, rkey, created_at, model, input, output, cache_read, cache_write from scrobble_usage order by created_at", )?; let rows = stmt.query_map([], |row| { Ok(crate::burn::UsageRow { did: row.get(0)?, rkey: row.get(1)?, created_at: row.get(2)?, model: row.get(3)?, tokens: [row.get(4)?, row.get(5)?, row.get(6)?, row.get(7)?], }) })?; Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?) }}
#[derive(Debug, Clone)]pub struct DailyRow { pub author_did: String, pub authored_at: String, pub repo_key: String,}
#[derive(Debug, Clone)]pub struct RecentCommit { pub sha: String, pub author_did: String, pub authored_at: String, pub message: String, pub repo_key: String, pub url: String,}
#[derive(Debug, Clone, Copy, Default)]pub struct Coverage { pub repos_total: u64, pub repos_scanned: u64, pub repos_failed: u64, pub branches_total: u64, pub branches_indexed: u64,}
/// Per day, how many logical changes each of [`COUNTED_COLUMNS`] counts,/// after merging one or more exports, with each export's `meta`, the/// repositories held for review and the evidence the exports carried.#[derive(Debug, Default)]pub struct MergedCounts { pub days: [Vec<(String, u64)>; 6], pub meta: Vec<HashMap<String, String>>, pub held: Vec<crate::rules::HeldRepo>, /// Per day, changes only held repositories have: default branch, every /// branch. pub held_days: [Vec<(String, u64)>; 2], pub evidence: Vec<crate::rules::RepoVerdict>, /// The largest repositories by counted commits, whatever their verdict. pub top: Vec<crate::rules::RepoVerdict>,}
/// The day columns of a counted change, for building statements.fn day_columns() -> String { COUNTED_COLUMNS.join(", ")}
/// Verdict reasons a manual rule gives. Those repos are listed by the/// excludes file, a fork record or their description, so a summary needn't/// list them again.const MANUAL_REASONS: &str = "('excludes_file', 'fork_record', 'described_copy')";
/// How many of the largest repositories, whatever their verdict, a summary/// lists with their evidence.pub const TOP_REPOS: usize = 50;
#[allow(clippy::too_many_arguments)]fn write_counted_rows( conn: &Connection, owner: &[String], agents: &[String], rules: &crate::rules::RuleSet, cap: &str, today: chrono::NaiveDate, shard: Shard, meta: &[(&str, String)],) -> Result<u64> { let columns = COUNTED_COLUMNS .map(|column| format!("{column} text")) .join(", "); let names = day_columns(); conn.execute_batch(&format!( "create table counts.meta (key text primary key, value text not null); create table counts.counted (key text primary key, {columns}) without rowid; create table counts.held_rows (repo text not null, key text not null, {columns}, primary key (repo, key)) without rowid; create table counts.candidates (repo text primary key, reason text not null, name text not null, url text not null, signals text not null); create table counts.evidence (repo text primary key, verdict text not null, reason text not null, name text not null, url text not null, commits integer, authors integer, tangled_commits integer, pre_creation integer, pre_creation_authors integer, history_before integer, signals text not null); create table counts.top as select * from counts.evidence where false; drop table if exists temp.counted_owner; drop table if exists temp.counted_agent; drop table if exists temp.counted_shard; drop table if exists temp.repo_rows; drop table if exists temp.counted_hold; create temp table counted_owner (did text primary key); create temp table counted_agent (did text primary key); create temp table counted_shard (repo_key text primary key); create temp table repo_rows (repo text not null, k text not null, verdict text not null, {columns}, primary key (repo, k)) without rowid; create temp table counted_hold (repo text primary key, reason text not null);" ))?; for (table, dids) in [("counted_owner", owner), ("counted_agent", agents)] { let mut stmt = conn.prepare(&format!("insert or ignore into temp.{table} values (?1)"))?; for did in dids { stmt.execute([did])?; } } let keys: Vec<String> = conn .prepare("select distinct repo_key from commit_refs where source = 'tangled'")? .query_map([], |row| row.get(0))? .collect::<rusqlite::Result<_>>()?; { let mut stmt = conn.prepare("insert into temp.counted_shard values (?1)")?; for key in keys.iter().filter(|key| shard.contains_repo_key(key)) { stmt.execute([key])?; } } let filter = if shard.is_sharded() { "and r.repo_key in (select repo_key from temp.counted_shard)" } else { "" }; conn.execute( &format!( "insert into temp.repo_rows {}", REPO_CHANGES_SQL.replace("{shard}", filter) ), params![ rules.grace_modifier(), cap, rules.params.collapse_rebased as i64 ], )?;
// Set aside every repository held by its verdict, and every one large // enough to be over the share threshold of the merged count: its count // in a window over the threshold of this export's own total is a // necessary condition, since the merged total is at least as large. Half // the threshold leaves room for the merge's windows ending a day apart. conn.execute( "insert into temp.counted_hold select distinct w.repo, e.reason from temp.repo_rows w join repo_evidence e on e.repo_did = w.repo where w.verdict = 'hold'", [], )?; let windows = crate::rules::windows(today, &rules.params.imported_before); let since: Vec<&str> = windows.iter().map(|(_, day)| day.as_str()).collect(); let totals: [u64; 4] = conn.query_row( "select coalesce(sum(d >= ?1), 0), coalesce(sum(d >= ?2), 0), coalesce(sum(d >= ?3), 0), coalesce(sum(d >= ?4), 0) from (select min(all_any) as d from temp.repo_rows where verdict != 'hold' group by k)", params![since[0], since[1], since[2], since[3]], |row| Ok([row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?]), )?; let per_repo: Vec<(String, [u64; 4])> = conn .prepare( "select repo, sum(all_any >= ?1), sum(all_any >= ?2), sum(all_any >= ?3), sum(all_any >= ?4) from temp.repo_rows where verdict != 'hold' group by repo", )? .query_map(params![since[0], since[1], since[2], since[3]], |row| { Ok(( row.get(0)?, [row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?], )) })? .collect::<rusqlite::Result<_>>()?; { let half = rules.params.hold_share_pct / 2.0; let mut stmt = conn.prepare("insert or ignore into temp.counted_hold values (?1, 'share')")?; for (repo, counts) in &per_repo { let large = counts.iter().zip(totals).any(|(count, total)| { *count >= rules.params.hold_share_min_changes && *count as f64 * 100.0 > half * total as f64 }); if large { stmt.execute([repo])?; } } }
let written = conn.execute( &format!( "insert into counts.counted select k, {mins} from temp.repo_rows where repo not in (select repo from temp.counted_hold) group by k", mins = COUNTED_COLUMNS .map(|column| format!("min({column})")) .join(", ") ), [], )?; conn.execute_batch(&format!( "insert into counts.held_rows select repo, k, {names} from temp.repo_rows where repo in (select repo from temp.counted_hold); insert into counts.candidates select h.repo, h.reason, coalesce(p.name, ''), coalesce(p.url, ''), coalesce(e.signals, '') from temp.counted_hold h left join repos p on p.repo_did = h.repo left join repo_evidence e on e.repo_did = h.repo; insert into counts.evidence select e.repo_did, e.verdict, e.reason, coalesce(p.name, ''), coalesce(p.url, ''), e.commits, e.authors, e.tangled_commits, e.pre_creation, e.pre_creation_authors, e.history_before, e.signals from repo_evidence e left join repos p on p.repo_did = e.repo_did where e.commits > 0 and e.verdict in ('exclude', 'hold') and e.reason not in {MANUAL_REASONS} and 'tangled:' || e.repo_did in (select repo_key from temp.counted_shard); insert into counts.top select e.repo_did, e.verdict, e.reason, coalesce(p.name, ''), coalesce(p.url, ''), e.commits, e.authors, e.tangled_commits, e.pre_creation, e.pre_creation_authors, e.history_before, e.signals from repo_evidence e left join repos p on p.repo_did = e.repo_did where e.commits > 0 and 'tangled:' || e.repo_did in (select repo_key from temp.counted_shard) order by e.commits desc, e.repo_did limit {TOP_REPOS};" ))?;
let mut stmt = conn.prepare("insert into counts.meta (key, value) values (?1, ?2)")?; for (key, value) in meta { stmt.execute(params![key, value])?; } // How many of this shard's repositories, with how many counted commits, // ended in each verdict. let mut tallies: std::collections::BTreeMap<String, (u64, u64)> = Default::default(); { let mut rows = conn.prepare("select repo_did, verdict, reason, commits from repo_evidence")?; let mut rows = rows.query([])?; while let Some(row) = rows.next()? { let repo: String = row.get(0)?; if !shard.contains(&repo) { continue; } let verdict: String = row.get(1)?; let reason: String = row.get(2)?; let tally = tallies.entry(format!("{verdict}:{reason}")).or_default(); tally.0 += 1; tally.1 += row.get::<_, i64>(3)? as u64; } } for (key, (repos, commits)) in tallies { stmt.execute(params![ format!("tally:{key}"), format!("{repos} {commits}") ])?; } stmt.execute(params!["rules_hash", rules.hash()])?; stmt.execute(params!["rules_version", crate::rules::RULES_VERSION])?; drop(stmt); conn.execute_batch( "drop table temp.counted_owner; drop table temp.counted_agent; drop table temp.counted_shard; drop table temp.repo_rows; drop table temp.counted_hold;", )?; Ok(written as u64)}
/// Merge counted-change exports written by [`Db::write_counted`]. A change/// in several exports (its repositories on different shards) counts once, on/// the earliest day any of them gives it, column by column.////// Each export sets aside the repositories that may need holding. Here, with/// every export's count, one over `policy`'s share of a window (the windows/// ending on the oldest export's day) is held unless reviewed, as is every/// one its verdict held; the rest count after all.pub fn merge_counted( paths: &[impl AsRef<Path>], policy: &crate::rules::HoldPolicy,) -> Result<MergedCounts> { let conn = Connection::open_in_memory()?; let columns = COUNTED_COLUMNS .map(|column| format!("{column} text")) .join(", "); conn.execute_batch(&format!( "create table merged (key text primary key, {columns}) without rowid; create table cand (repo text not null, key text not null, {columns}, primary key (repo, key)) without rowid; create table cand_meta (repo text primary key, reason text not null, name text not null, url text not null, signals text not null); create table evid (repo text primary key, verdict text not null, reason text not null, name text not null, url text not null, commits integer, authors integer, tangled_commits integer, pre_creation integer, pre_creation_authors integer, history_before integer, signals text not null); create table top as select * from evid where false;" ))?; let earliest = COUNTED_COLUMNS .map(|column| { // min() of two values is null if either is; keep whichever exists. format!( "{column} = coalesce(min({column}, excluded.{column}), {column}, excluded.{column})" ) }) .join(", "); let names = day_columns(); let mut merged = MergedCounts::default(); for path in paths { let path = path.as_ref(); if !path.exists() { anyhow::bail!("no counted changes at {}", path.display()); } conn.execute( "attach database ?1 as export", [path.to_string_lossy().as_ref()], )?; let read = (|| -> Result<HashMap<String, String>> { conn.execute( &format!( "insert into merged (key, {names}) select key, {names} from export.counted where true on conflict (key) do update set {earliest}" ), [], )?; conn.execute_batch(&format!( "insert or ignore into cand (repo, key, {names}) select repo, key, {names} from export.held_rows; insert or ignore into cand_meta select * from export.candidates; insert or ignore into evid select * from export.evidence; insert into top select * from export.top;" ))?; let meta = conn .prepare("select key, value from export.meta")? .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))? .collect::<rusqlite::Result<_>>()?; Ok(meta) })(); let detached = conn.execute("detach database export", []); let meta = read.with_context(|| format!("merging {}", path.display()))?; detached?; merged.meta.push(meta); }
let end = merged .meta .iter() .filter_map(|meta| meta.get("generated_at")) .min() .and_then(|at| at.get(..10)) .and_then(|day| chrono::NaiveDate::parse_from_str(day, "%Y-%m-%d").ok()) .unwrap_or_else(|| Utc::now().date_naive()); let windows = crate::rules::windows(end, &policy.imported_before); let since: Vec<&str> = windows.iter().map(|(_, day)| day.as_str()).collect(); // Shares against everything that could count: the merged changes and // every repository set aside for its size. let totals: [u64; 4] = conn.query_row( "select coalesce(sum(d >= ?1), 0), coalesce(sum(d >= ?2), 0), coalesce(sum(d >= ?3), 0), coalesce(sum(d >= ?4), 0) from (select key, min(d) as d from ( select key, all_any as d from merged union all select key, all_any from cand where repo in (select repo from cand_meta where reason = 'share')) group by key)", params![since[0], since[1], since[2], since[3]], |row| Ok([row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?]), )?; let candidates: Vec<(String, String, String, String, String)> = conn .prepare("select repo, reason, name, url, signals from cand_meta order by repo")? .query_map([], |row| { Ok(( row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?, )) })? .collect::<rusqlite::Result<_>>()?; let mut held = Vec::new(); for (repo, reason, name, url, signals) in candidates { let counts: [u64; 4] = conn.query_row( "select coalesce(sum(all_any >= ?2), 0), coalesce(sum(all_any >= ?3), 0), coalesce(sum(all_any >= ?4), 0), coalesce(sum(all_any >= ?5), 0) from cand where repo = ?1", params![repo, since[0], since[1], since[2], since[3]], |row| Ok([row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?]), )?; let share = counts .iter() .zip(totals) .filter(|(count, _)| **count >= policy.min_changes) .map(|(count, total)| *count as f64 * 100.0 / total.max(1) as f64) .fold(0.0f64, f64::max); let hold = if reason == "share" { share > policy.share_pct && !policy.reviewed.contains(&repo) } else { true }; if hold { held.push((repo, reason, name, url, signals, share)); } else { conn.execute( &format!( "insert into merged (key, {names}) select key, {names} from cand where repo = ?1 and true on conflict (key) do update set {earliest}" ), [&repo], )?; } } // What each held repository would add, and per day what only held ones // have. for (repo, reason, name, url, signals, share) in held { let mut changes = std::collections::BTreeMap::new(); let mut changes_all = std::collections::BTreeMap::new(); for (label, first) in &windows { let (default, any): (u64, u64) = conn.query_row( "select coalesce(sum(all_default >= ?2), 0), coalesce(sum(all_any >= ?2), 0) from cand where repo = ?1 and key not in (select key from merged)", params![repo, first], |row| Ok((row.get(0)?, row.get(1)?)), )?; changes.insert(label.to_string(), default); changes_all.insert(label.to_string(), any); } merged.held.push(crate::rules::HeldRepo { share_pct: (reason == "share").then(|| (share * 100.0).round() / 100.0), repo, changes: changes.get("all").copied().unwrap_or(0), name, url, reason, changes_by_window: changes, changes_all_branches: changes_all, signals: split_signals(&signals), }); } let held_list: Vec<String> = merged.held.iter().map(|h| h.repo.clone()).collect(); conn.execute_batch("create temp table held (repo text primary key);")?; { let mut stmt = conn.prepare("insert into temp.held values (?1)")?; for repo in &held_list { stmt.execute([repo])?; } } for (index, column) in ["all_default", "all_any"].iter().enumerate() { merged.held_days[index] = conn .prepare(&format!( "select d, count(*) from ( select key, min({column}) as d from cand where repo in (select repo from temp.held) and key not in (select key from merged) group by key) where d is not null group by d order by d" ))? .query_map([], |row| Ok((row.get(0)?, row.get::<_, i64>(1)? as u64)))? .collect::<rusqlite::Result<_>>()?; } let verdict_row = |row: &rusqlite::Row<'_>| -> rusqlite::Result<crate::rules::RepoVerdict> { Ok(crate::rules::RepoVerdict { repo_did: row.get(0)?, verdict: row.get(1)?, reason: row.get(2)?, name: row.get(3)?, url: row.get(4)?, commits: row.get::<_, i64>(5)? as u64, authors: row.get::<_, i64>(6)? as u64, tangled_commits: row.get::<_, i64>(7)? as u64, pre_creation: row.get::<_, i64>(8)? as u64, pre_creation_authors: row.get::<_, i64>(9)? as u64, history_before: row.get::<_, i64>(10)? as u64, signals: split_signals(&row.get::<_, String>(11)?), }) }; let columns = "repo, verdict, reason, name, url, commits, authors, tangled_commits, pre_creation, pre_creation_authors, history_before, signals"; merged.evidence = conn .prepare(&format!( "select {columns} from evid order by commits desc, repo" ))? .query_map([], verdict_row)? .collect::<rusqlite::Result<_>>()?; merged.top = conn .prepare(&format!( "select {columns} from top order by commits desc, repo limit {TOP_REPOS}" ))? .query_map([], verdict_row)? .collect::<rusqlite::Result<_>>()?;
for (index, column) in COUNTED_COLUMNS.iter().enumerate() { merged.days[index] = conn .prepare(&format!( "select {column}, count(*) from merged where {column} is not null group by {column} order by {column}" ))? .query_map([], |row| Ok((row.get(0)?, row.get::<_, i64>(1)? as u64)))? .collect::<rusqlite::Result<_>>()?; } Ok(merged)}
fn split_signals(signals: &str) -> Vec<String> { signals .split(',') .filter(|signal| !signal.is_empty()) .map(str::to_string) .collect()}
/// Copy the seed's rows that belong to `shard` into the main database.fn seed_shard_rows(conn: &Connection, shard: Shard) -> Result<()> { conn.execute_batch( "drop table if exists temp.seed_shard; create temp table seed_shard (repo_did text primary key);", )?; let repos: Vec<String> = conn .prepare( "select repo_did from seed.repos union select substr(repo_key, 9) from seed.commit_refs where source = 'tangled'", )? .query_map([], |row| row.get(0))? .collect::<rusqlite::Result<_>>()?; let mut stmt = conn.prepare("insert or ignore into temp.seed_shard values (?1)")?; for repo in repos.iter().filter(|repo| shard.contains(repo)) { stmt.execute([repo])?; } drop(stmt); let in_shard = "(select repo_did from temp.seed_shard)"; conn.execute_batch("begin")?; let copied = (|| -> Result<()> { for table in ["meta", "identities", "repos", "github_repos"] { copy_seed_table(conn, table, "")?; } copy_seed_table(conn, "branches", &format!("where repo_did in {in_shard}"))?; copy_seed_table( conn, "commit_refs", &format!("where source = 'tangled' and substr(repo_key, 9) in {in_shard}"), )?; copy_seed_table( conn, "commits", "where sha in (select sha from main.commit_refs)", )?; Ok(()) })(); match copied { Ok(()) => conn.execute_batch("commit; drop table temp.seed_shard;")?, Err(err) => { let _ = conn.execute_batch("rollback"); return Err(err); } } Ok(())}
/// `insert or ignore` the seed's rows of `table` into the main database by/// column name: the two may have gained columns in a different order.fn copy_seed_table(conn: &Connection, table: &str, filter: &str) -> Result<()> { let columns: Vec<String> = conn .prepare(&format!( "select name from pragma_table_info('{table}', 'main')" ))? .query_map([], |row| row.get(0))? .collect::<rusqlite::Result<_>>()?; let columns = columns.join(", "); conn.execute( &format!( "insert or ignore into main.{table} ({columns}) select {columns} from seed.{table} {filter}" ), [], ) .with_context(|| format!("copying {table} from the seed"))?; Ok(())}
pub fn now_string() -> String { Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)}
/// Parse an arbitrary RFC 3339 stamp into the canonical UTC form used for/// lexicographic comparisons in SQLite.pub fn canonical_time(raw: &str) -> Option<String> { DateTime::parse_from_rfc3339(raw).ok().map(|stamp| { stamp .with_timezone(&Utc) .to_rfc3339_opts(chrono::SecondsFormat::Secs, true) })}
#[cfg(test)]mod tests { use super::*;
fn commit(sha: &str, author: &str) -> Commit { Commit { sha: sha.to_string(), change_id: String::new(), author_did: author.to_string(), author_email: author.to_string(), author_name: String::new(), authored_at: "2026-10-20T12:00:00Z".to_string(), committed_at: "2026-10-20T12:00:00Z".to_string(), message: "x".to_string(), } }
fn reference(sha: &str, repo: &str) -> CommitRef { CommitRef { sha: sha.to_string(), source: "tangled".to_string(), repo_key: repo.to_string(), url: String::new(), on_default: true, } }
#[test] fn a_run_of_failures_is_kept_until_a_read_succeeds() { let db = Db::in_memory().unwrap(); db.conn .lock() .unwrap() .execute_batch( "insert into repos (repo_did, first_seen, last_seen) values ('did:plc:r', '', '')", ) .unwrap(); let state = |db: &Db| -> (i64, Option<String>, Option<String>) { db.conn .lock() .unwrap() .query_row( "select fail_count, failing_since, last_success from repos", [], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), ) .unwrap() }; db.mark_repo_scanned("did:plc:r", Some("returned 502")) .unwrap(); let (count, since, success) = state(&db); assert_eq!((count, success), (1, None)); db.conn .lock() .unwrap() .execute_batch("update repos set failing_since = '2026-10-01T00:00:00Z'") .unwrap(); db.mark_repo_scanned("did:plc:r", Some("returned 502")) .unwrap(); // The run keeps its start. assert_eq!(state(&db).0, 2); assert_eq!(state(&db).1.as_deref(), Some("2026-10-01T00:00:00Z")); assert!(since.is_some()); db.mark_repo_scanned("did:plc:r", None).unwrap(); let (count, since, success) = state(&db); assert_eq!((count, since), (0, None)); assert!(success.is_some()); }
#[test] fn a_database_from_before_failure_runs_starts_them_from_its_last_scan() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("old.sqlite3"); { let conn = Connection::open(&path).unwrap(); conn.execute_batch( "create table repos ( repo_did text primary key, owner_did text not null default '', name text not null default '', knot text not null default '', url text not null default '', first_seen text not null, last_seen text not null, last_scanned text, last_error text); insert into repos (repo_did, first_seen, last_seen, last_scanned, last_error) values ('did:plc:ok', '', '', '2026-10-22T08:00:00Z', null), ('did:plc:bad', '', '', '2026-10-25T08:00:00Z', 'returned 502'), ('did:plc:new', '', '', null, null);", ) .unwrap(); } let db = Db::open(&path).unwrap(); let rows: Vec<(String, Option<String>, i64, Option<String>)> = db .conn .lock() .unwrap() .prepare( "select repo_did, last_success, fail_count, failing_since from repos order by repo_did", ) .unwrap() .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))) .unwrap() .collect::<rusqlite::Result<_>>() .unwrap(); assert_eq!( rows, [ ( "did:plc:bad".into(), None, 1, Some("2026-10-25T08:00:00Z".into()) ), ("did:plc:new".into(), None, 0, None), ( "did:plc:ok".into(), Some("2026-10-22T08:00:00Z".into()), 0, None ), ] ); }
#[test] fn run_coverage_counts_this_runs_reads_and_estimates_what_is_missing() { let db = Db::in_memory().unwrap(); let now: DateTime<Utc> = "2026-10-26T00:00:00Z".parse().unwrap(); db.conn .lock() .unwrap() .execute_batch( "insert into repos (repo_did, knot, first_seen, last_seen, last_scanned, last_success, last_error, fail_count, failing_since, excluded, discovered_via) values ('did:plc:fresh', 'knot1.tangled.sh', '', '', '2026-10-25T12:00:00Z', '2026-10-25T12:00:00Z', null, 0, null, 0, 'search'), ('did:plc:stale', 'knot1.tangled.sh', '', '', '2026-10-21T00:00:00Z', '2026-10-21T00:00:00Z', null, 0, null, 0, 'search'), ('did:plc:down', 'git.jcs.org', '', '', '2026-10-25T13:00:00Z', null, 'GET x returned 502: bad gateway', 1, '2026-10-25T13:00:00Z', 0, 'search'), ('did:plc:mine', 'localhost:6444', '', '', null, null, null, 0, null, 0, 'search'), ('did:plc:owner/old', 'knot1.tangled.sh', '', '', '2026-10-25T14:00:00Z', '2026-10-25T14:00:00Z', null, 0, null, 0, 'search_legacy'), ('did:plc:fork', 'knot1.tangled.sh', '', '', null, null, null, 0, null, 1, 'search');", ) .unwrap(); // The stale repository made 60 changes in the 30 days before its last // read on 10-21: 2 a day, so about 10 in the 5 days since. for day in 0..60 { let mut change = commit(&format!("s{day}"), "did:plc:x"); let at = (now - chrono::Duration::days(5) - chrono::Duration::hours(12 * day + 1)) .to_rfc3339_opts(chrono::SecondsFormat::Secs, true); change.authored_at = at.clone(); change.committed_at = at; db.insert_commit( &change, &reference(&format!("s{day}"), "tangled:did:plc:stale"), ) .unwrap(); } // The failing one was never read here; its newest stored change, a // day ago, stands in for its last read. let mut change = commit("d", "did:plc:x"); change.authored_at = "2026-10-25T00:00:00Z".into(); db.insert_commit(&change, &reference("d", "tangled:did:plc:down")) .unwrap();
let coverage = db .run_coverage(Shard::WHOLE, "2026-10-25T00:00:00Z", now) .unwrap(); assert_eq!(coverage.repos_discovered, 6); assert_eq!(coverage.repos_excluded, 1); assert_eq!(coverage.repos_unreachable, 1); assert_eq!(coverage.unreachable["private_knot"], 1); assert_eq!(coverage.repos_reachable, 4); assert_eq!(coverage.repos_scanned, 2); assert_eq!(coverage.repos_failed, 1); assert_eq!(coverage.failed["http_5xx"], 1); assert_eq!(coverage.repos_not_attempted, 1); assert_eq!(coverage.discovered_via["search"], 4); assert_eq!(coverage.discovered_via["search_legacy"], 1); assert_eq!(coverage.pct(), 50.0); // 60 changes over 30 days, 5 unread days: 10. The failing one's one // change over 30 days, one unread day: nothing to round up. assert_eq!(coverage.missing_changes_30d_est, 10); assert_eq!(coverage.missing_unknown_repos, 0); assert_eq!(coverage.unreachable_changes_30d_est, 0); }
#[test] fn unfinished_and_unmarked_history_needs_a_walk() { let db = Db::in_memory().unwrap(); let mut side = reference("s", "tangled:did:plc:unmarked"); side.on_default = false; db.insert_commit(&commit("s", "did:plc:me"), &side).unwrap(); db.upsert_branch("did:plc:unmarked", "main", "s", true) .unwrap(); db.insert_commit( &commit("d", "did:plc:me"), &reference("d", "tangled:did:plc:done"), ) .unwrap(); db.upsert_branch("did:plc:done", "main", "d", true).unwrap(); db.upsert_branch("did:plc:requeued", "main", "x", false) .unwrap(); db.upsert_branch("did:plc:capped", "main", "y", false) .unwrap(); db.conn .lock() .unwrap() .execute_batch( "insert into repos (repo_did, first_seen, last_seen, last_error) values ('did:plc:capped', '', '', 'truncated after 41 pages')", ) .unwrap(); let needing = db.repos_needing_walk().unwrap(); let mut needing: Vec<_> = needing.iter().map(String::as_str).collect(); needing.sort(); assert_eq!(needing, ["did:plc:requeued", "did:plc:unmarked"]);
// A deleted branch that was never indexed stops holding its repo. db.upsert_branch("did:plc:requeued", "gone", "z", false) .unwrap(); assert_eq!(db.prune_branches("did:plc:requeued", &["main"]).unwrap(), 1); db.upsert_branch("did:plc:requeued", "main", "x", true) .unwrap(); assert!( !db.repos_needing_walk() .unwrap() .contains("did:plc:requeued") ); }
#[test] fn same_commit_on_many_branches_counts_once() { let db = Db::in_memory().unwrap(); db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r1")) .unwrap(); db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r1")) .unwrap(); db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r1")) .unwrap(); assert_eq!(db.tangled_total().unwrap(), 1); assert_eq!( db.commits_for_dids(&["did:plc:me".into()], "tangled") .unwrap(), 1 ); }
#[test] fn same_sha_in_two_repos_counts_once_and_keeps_both_refs() { let db = Db::in_memory().unwrap(); db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r1")) .unwrap(); db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r2")) .unwrap(); assert_eq!(db.tangled_total().unwrap(), 1); assert_eq!(db.top_repos(10).unwrap().len(), 2); }
#[test] fn momentum_windows_are_half_open() { let db = Db::in_memory().unwrap(); let mut january = commit("a", "did:plc:me"); january.authored_at = "2026-01-01T00:00:00Z".into(); db.insert_commit(&january, &reference("a", "tangled:r1")) .unwrap(); let mut february = commit("b", "did:plc:me"); february.authored_at = "2026-02-01T00:00:00Z".into(); db.insert_commit(&february, &reference("b", "tangled:r1")) .unwrap(); assert_eq!( db.repo_commit_counts_between( "tangled:r1", &["did:plc:me".into()], "2026-01-15T00:00:00Z", "2026-02-15T00:00:00Z", ) .unwrap(), 1 ); assert_eq!( db.repo_commit_counts_between( "tangled:r1", &["did:plc:me".into()], "2026-01-01T00:00:00Z", "2026-02-01T00:00:00Z", ) .unwrap(), 1 ); }
#[test] fn recent_shows_named_personal_inbox_but_not_work_or_redacted() { let db = Db::in_memory().unwrap(); for (sha, repo_key) in [ ("a", "github:inbox:personal:me/public"), ("b", "github:inbox:work:me/secret"), ("c", "github:redacted:personal:h_private"), ] { db.insert_commit( &commit(sha, "github:personal"), &CommitRef { sha: sha.into(), source: "github".into(), repo_key: repo_key.into(), url: String::new(), on_default: true, }, ) .unwrap(); } let rows = db .recent_commits_for_dids_all(&["github:personal".into()], 10) .unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].repo_key, "github:inbox:personal:me/public"); }
#[test] fn global_counts_find_repositories_by_primary_key() { // A scan of `repos` per commit reference made these queries take hours // on a real global index; each lookup must be a primary-key search. let db = Db::in_memory().unwrap(); let conn = db.conn.lock().unwrap(); conn.execute_batch( "create temp table counted_owner (did text primary key); create temp table counted_agent (did text primary key);", ) .unwrap(); let counted = REPO_CHANGES_SQL.replace("{shard}", ""); for sql in [TANGLED_DAILY_COUNTS_SQL, REPO_LOGICAL_COUNTS_SQL, &counted] { let mut stmt = conn.prepare(&format!("explain query plan {sql}")).unwrap(); let params = std::iter::repeat_n("-30 days", stmt.parameter_count()); let plan: Vec<String> = stmt .query_map(rusqlite::params_from_iter(params), |row| { row.get::<_, String>(3) }) .unwrap() .collect::<rusqlite::Result<_>>() .unwrap(); assert!( !plan.iter().any(|step| step == "SCAN p"), "a repos subquery scans the whole table: {plan:?}" ); assert!( plan.iter().any(|step| step.starts_with("SEARCH p USING")), "no repos subquery uses an index: {plan:?}" ); } }
#[test] fn the_branch_scope_decides_which_tangled_commits_count() { let db = Db::in_memory().unwrap(); let side = |sha: &str| CommitRef { on_default: false, ..reference(sha, "tangled:r1") }; db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r1")) .unwrap(); db.insert_commit(&commit("b", "did:plc:me"), &side("b")) .unwrap(); let github = CommitRef { source: "github".into(), repo_key: "github:me/x".into(), on_default: false, ..reference("g", "") }; db.insert_commit(&commit("g", "did:plc:me"), &github) .unwrap(); let me = ["did:plc:me".to_string()]; assert_eq!( db.count_logical_commits(&me, Some("tangled"), None) .unwrap(), 1 ); // A GitHub reference from a side branch is left out the same way. assert_eq!(db.count_logical_commits(&me, None, None).unwrap(), 1); db.set_branch_scope(BranchScope::All).unwrap(); assert_eq!(db.branch_scope(), BranchScope::All); assert_eq!( db.count_logical_commits(&me, Some("tangled"), None) .unwrap(), 2 ); assert_eq!(db.count_logical_commits(&me, None, None).unwrap(), 3); }
#[test] fn a_default_branch_sighting_upgrades_a_side_branch_one() { let db = Db::in_memory().unwrap(); let me = ["did:plc:me".to_string()]; let side = CommitRef { on_default: false, ..reference("a", "tangled:did:plc:r1") }; db.insert_commit(&commit("a", "did:plc:me"), &side).unwrap(); assert_eq!( db.count_logical_commits(&me, Some("tangled"), None) .unwrap(), 0 ); db.insert_commit( &commit("a", "did:plc:me"), &reference("a", "tangled:did:plc:r1"), ) .unwrap(); assert_eq!( db.count_logical_commits(&me, Some("tangled"), None) .unwrap(), 1 ); db.insert_commit(&commit("a", "did:plc:me"), &side).unwrap(); assert_eq!( db.count_logical_commits(&me, Some("tangled"), None) .unwrap(), 1 ); }
#[test] fn the_default_branch_walk_marks_what_side_branches_stored() { let db = Db::in_memory().unwrap(); let side = |sha: &str| CommitRef { on_default: false, ..reference(sha, "tangled:did:plc:r1") }; for sha in ["a", "b"] { db.insert_commit(&commit(sha, "did:plc:me"), &side(sha)) .unwrap(); } assert!(db.repo_default_shas("did:plc:r1").unwrap().is_empty()); db.mark_on_default("did:plc:r1", &["a".to_string()]) .unwrap(); let marked = db.repo_default_shas("did:plc:r1").unwrap(); assert!(marked.contains("a") && !marked.contains("b")); assert_eq!(db.repo_commit_shas("did:plc:r1").unwrap().len(), 2); }
#[test] fn adding_the_branch_marker_requeues_branches_and_keeps_github_counting() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("old.sqlite3"); { let conn = Connection::open(&path).unwrap(); conn.execute_batch( "create table commit_refs ( sha text not null, source text not null, repo_key text not null, url text not null default '', primary key (sha, source, repo_key)); create table branches ( repo_did text not null, name text not null, head_sha text not null default '', indexed integer not null default 0, last_scanned text, primary key (repo_did, name)); insert into branches values ('did:plc:r1', 'main', 'abc', 1, null); insert into commit_refs values ('abc', 'tangled', 'tangled:did:plc:r1', ''); insert into commit_refs values ('def', 'github', 'github:me/x', '');", ) .unwrap(); } let db = Db::open(&path).unwrap(); let conn = db.conn.lock().unwrap(); let indexed: i64 = conn .query_row("select indexed from branches", [], |row| row.get(0)) .unwrap(); let marks: Vec<(String, i64)> = conn .prepare("select source, on_default from commit_refs order by source") .unwrap() .query_map([], |row| Ok((row.get(0)?, row.get(1)?))) .unwrap() .collect::<rusqlite::Result<_>>() .unwrap(); assert_eq!(indexed, 0); assert_eq!( marks, [("github".to_string(), 1), ("tangled".to_string(), 0)] ); }
#[test] fn a_login_named_personal_github_identity_folds_into_the_current_one() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("old.sqlite3"); { let db = Db::open(&path).unwrap(); let conn = db.conn.lock().unwrap(); conn.execute_batch( "insert into identities (did, handle, kind, color, last_seen) values ('github:octocat', 'octocat', 'github', 'series-6', 'x'), ('github-work:redacted', 'work GitHub', 'github', 'series-7', 'x'); insert into commits (sha, author_did, authored_at, committed_at, first_source) values ('a', 'github:octocat', 't', 't', 'github'), ('b', 'github-work:redacted', 't', 't', 'github');", ) .unwrap(); } let db = Db::open(&path).unwrap(); let conn = db.conn.lock().unwrap(); let dids: Vec<String> = conn .prepare("select did from identities order by did") .unwrap() .query_map([], |row| row.get(0)) .unwrap() .collect::<rusqlite::Result<_>>() .unwrap(); assert_eq!(dids, ["github-work:redacted", "github:personal"]); let authors: Vec<String> = conn .prepare("select author_did from commits order by sha") .unwrap() .query_map([], |row| row.get(0)) .unwrap() .collect::<rusqlite::Result<_>>() .unwrap(); assert_eq!(authors, ["github:personal", "github-work:redacted"]); }
#[test] fn excluded_repos_leave_the_denominator() { let db = Db::in_memory().unwrap(); db.upsert_repo(&Repo { repo_did: "did:plc:repo".into(), owner_did: "did:plc:me".into(), name: "example".into(), knot: "knot.test".into(), url: "https://tangled.org/me/example".into(), website: None, fork: true, created_at: None, active: true, last_scanned: None, last_error: None, }) .unwrap(); assert_eq!(db.repo_coverage().unwrap().repos_total, 1); db.exclude_repo("did:plc:repo", "fork").unwrap(); assert_eq!(db.repo_coverage().unwrap().repos_total, 0); }
#[test] fn a_flushed_database_file_is_whole_without_its_log() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("live.sqlite3"); let db = Db::open(&path).unwrap(); db.insert_commit(&commit("a", "did:plc:me"), &reference("a", "tangled:r1")) .unwrap(); db.flush_wal().unwrap(); // What an upload of the file alone carries. let copy = dir.path().join("uploaded.sqlite3"); std::fs::copy(&path, ©).unwrap(); assert_eq!(Db::open(©).unwrap().tangled_total().unwrap(), 1); }
#[test] fn excluded_repositories_leave_the_global_daily_counts() { let db = Db::in_memory().unwrap(); for (did, name) in [("did:plc:keep", "keep"), ("did:plc:drop", "drop")] { db.upsert_repo(&Repo { repo_did: did.into(), owner_did: "did:plc:me".into(), name: name.into(), knot: "knot.test".into(), url: format!("https://tangled.org/me/{name}"), website: None, fork: false, created_at: None, active: true, last_scanned: None, last_error: None, }) .unwrap(); } db.insert_commit( &commit("a", "did:plc:me"), &reference("a", "tangled:did:plc:keep"), ) .unwrap(); db.insert_commit( &commit("b", "did:plc:me"), &reference("b", "tangled:did:plc:drop"), ) .unwrap(); db.exclude_repo("did:plc:drop", "oversized").unwrap(); assert_eq!( db.tangled_daily_counts(0).unwrap(), vec![("2026-10-20".to_string(), 1)] ); }
#[test] fn global_daily_counts_collapse_by_change_id_and_day() { let db = Db::in_memory().unwrap();
let mut original = commit("a", "did:plc:me"); original.change_id = "Iabc".into(); original.authored_at = "2026-01-01T00:00:00Z".into(); db.insert_commit(&original, &reference("a", "tangled:r1")) .unwrap();
let mut rebased = commit("b", "did:plc:me"); rebased.change_id = "Iabc".into(); rebased.authored_at = "2026-02-01T00:00:00Z".into(); db.insert_commit(&rebased, &reference("b", "tangled:r1")) .unwrap();
let mut same_day = commit("c", "did:plc:me"); same_day.authored_at = "2026-01-01T18:00:00Z".into(); db.insert_commit(&same_day, &reference("c", "tangled:r1")) .unwrap();
let mut github = commit("d", "github:personal"); github.authored_at = "2026-01-01T00:00:00Z".into(); db.insert_commit( &github, &CommitRef { sha: "d".into(), source: "github".into(), repo_key: "github:personal/repo".into(), url: String::new(), on_default: true, }, ) .unwrap();
assert_eq!( db.tangled_daily_counts(0).unwrap(), vec![("2026-01-01".to_string(), 2)] ); }
#[test] fn later_identity_resolution_upgrades_a_commit_without_duplicating_it() { let db = Db::in_memory().unwrap(); let mut unknown = commit("a", ""); unknown.author_email = "someone@example.test".into(); db.insert_commit(&unknown, &reference("a", "tangled:r1")) .unwrap(); let mut known = commit("a", "did:plc:me"); known.author_email = "someone@example.test".into(); db.insert_commit(&known, &reference("a", "tangled:r1")) .unwrap(); assert_eq!(db.tangled_total().unwrap(), 1); assert_eq!( db.commits_for_dids(&["did:plc:me".into()], "tangled") .unwrap(), 1 ); }
#[test] fn rebased_commit_collapses_by_change_id() { let db = Db::in_memory().unwrap();
let mut original = commit("a", "did:plc:me"); original.change_id = "Iabc".into(); original.authored_at = "2026-01-01T00:00:00Z".into(); db.insert_commit(&original, &reference("a", "tangled:r1")) .unwrap();
let mut rebased = commit("b", "did:plc:me"); rebased.change_id = "Iabc".into(); rebased.authored_at = "2026-02-01T00:00:00Z".into(); db.insert_commit(&rebased, &reference("b", "tangled:r1")) .unwrap();
let plain = commit("c", "did:plc:me"); db.insert_commit(&plain, &reference("c", "tangled:r1")) .unwrap();
// Three unique Git objects, two logical changes. assert_eq!(db.tangled_total().unwrap(), 3); assert_eq!(db.count_commits(&[], Some("tangled"), None).unwrap(), 3); assert_eq!( db.count_logical_commits(&[], Some("tangled"), None) .unwrap(), 2 ); let counts = db.identity_counts_logical("tangled").unwrap(); assert_eq!(counts.get("did:plc:me"), Some(&2));
let rows = db.daily_rows_logical("2025-01-01T00:00:00Z").unwrap(); assert_eq!(rows.len(), 2); assert!(rows.iter().all(|row| row.repo_key == "tangled:r1")); assert!( rows.iter() .any(|row| row.authored_at == "2026-01-01T00:00:00Z") ); assert!( !rows .iter() .any(|row| row.authored_at == "2026-02-01T00:00:00Z") ); }
#[test] fn a_mirrored_change_counts_once_in_its_tangled_repository() { let db = Db::in_memory().unwrap(); let mut mirrored = commit("m", "did:plc:me"); mirrored.authored_at = "2026-01-01T00:00:00Z".into(); let github = CommitRef { source: "github".into(), ..reference("m", "github:me/mirror") }; db.insert_commit(&mirrored, &github).unwrap(); db.insert_commit(&mirrored, &reference("m", "tangled:r1")) .unwrap(); let rows = db.daily_rows_logical("2025-01-01T00:00:00Z").unwrap(); let repos: Vec<_> = rows.iter().map(|row| row.repo_key.as_str()).collect(); assert_eq!(repos, ["tangled:r1"]); }
#[test] fn canonical_time_normalises_offsets() { assert_eq!( canonical_time("2026-10-20T12:00:00-04:00").unwrap(), "2026-10-20T16:00:00Z" ); }}