Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
160 linesCodeBlameRaw
1//! The search service: one search across all of g1t, and Explore.
2//!
3//! - **What it searches.** Repositories (name, description, topics, the
4//! opening of the README), code on default branches (paths and
5//! contents), work items and pull requests (titles and bodies), people
6//! and workspaces. Everything public, and private content the viewer is
7//! a member of.
8//! - **The index.** D1 with FTS5, a database that costs nothing while no
9//! one uses it: words for prose (`unicode61`), every run of three
10//! characters for code (`trigram`), each table kept in step with its
11//! index by triggers. See `migrations/0001_init.sql`.
12//! - **Visibility.** Decided when a query runs, twice: in SQL against the
13//! viewer's memberships now, then on the page against the repos service.
14//! See `visibility.rs`.
15//! - **Keeping it current.** From the events bus (`g1t-events-search`) and
16//! its own job queue (`g1t-search-jobs`), with the cost of each push
17//! capped. See `index.rs`.
18//!
19//! Reached through service bindings: `POST /rpc/<method>`; see
20//! `g1t_contracts::search`.
21
22mod index;
23#[cfg(test)]
24mod migrated;
25mod query;
26mod read;
27mod rules;
28mod snippet;
29mod sql;
30mod store;
31mod visibility;
32
33use std::cell::RefCell;
34use std::collections::HashMap;
35
36use g1t_contracts::User;
37use g1t_contracts::billing::PauseLevel;
38use g1t_contracts::events::Event;
39use g1t_kit::{args, reply, rpc_method};
40use worker::{Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, event};
41
42use index::Job;
43
44/// The queue of this service's own jobs.
45const JOBS_QUEUE: &str = "g1t-search-jobs";
46
47/// How often an isolate looks for backfill pages parked while indexing
48/// was paused.
49const RESUME_EVERY_MS: u64 = 5 * 60 * 1000;
50
51thread_local! {
52 static RESUME_CHECKED: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
53}
54
55/// Whether to look for parked pages now; marks it looked for.
56fn resume_due(now: u64) -> bool {
57 RESUME_CHECKED.with(|last| {
58 if now.saturating_sub(last.get()) < RESUME_EVERY_MS {
59 return false;
60 }
61 last.set(now);
62 true
63 })
64}
65
66pub struct Search {
67 db: D1Database,
68 env: Env,
69 repos: Fetcher,
70 work: Fetcher,
71 identity: Fetcher,
72 /// Each workspace's own principal, asked of identity once per invocation.
73 actors: RefCell<HashMap<String, Option<User>>>,
74}
75
76impl Search {
77 fn new(env: Env) -> Result<Self> {
78 Ok(Search {
79 db: env.d1("DB")?,
80 repos: env.service("REPOS")?,
81 work: env.service("WORK")?,
82 identity: env.service("IDENTITY")?,
83 actors: RefCell::new(HashMap::new()),
84 env,
85 })
86 }
87}
88
89#[event(fetch)]
90async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
91 let Some(method) = rpc_method(&request) else {
92 return Response::error("Not found", 404);
93 };
94 // A replica near the caller when it asks for one (crates/kit/src/d1.rs).
95 let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
96 let mut search = Search::new(env)?;
97 search.db = db;
98 let body: serde_json::Value = request.json().await?;
99 let answered = match method.as_str() {
100 "search" => reply(&search.search(args(body)?).await?),
101 "suggest" => reply(&search.suggest(args(body)?).await?),
102 "explore" => reply(&search.explore(args(body)?).await?),
103 _ => Response::error("Unknown method", 404),
104 };
105 served.finish(answered)
106}
107
108/// Events from the bus, and this service's own jobs. Indexing is
109/// idempotent, so a message that fails is retried.
110#[event(queue)]
111async fn queue(batch: MessageBatch<serde_json::Value>, env: Env, _ctx: Context) -> Result<()> {
112 let search = Search::new(env)?;
113 let jobs = batch.queue() == JOBS_QUEUE;
114 // Indexing paused across g1t (billing's `platform_pause`, kept 30
115 // seconds in the isolate): backfills wait, and events still keep the
116 // index current. Without a billing binding nothing is ever paused.
117 let paused = match search.env.service("BILLING") {
118 Ok(billing) => g1t_kit::pause::paused(&billing, PauseLevel::Indexing).await,
119 Err(_) => false,
120 };
121 if !jobs && !paused {
122 if let Err(error) = search.ensure_backfill().await {
123 worker::console_error!("search: could not start the backfill: {error}");
124 }
125 // Pages parked while paused, looked for at most every few minutes.
126 if resume_due(g1t_kit::now_ms()) {
127 match search.resume_parked().await {
128 Ok(0) => {}
129 Ok(found) => worker::console_log!("search: resumed {found} parked backfill pages"),
130 Err(error) => worker::console_error!("search: could not resume parked backfill pages: {error}"),
131 }
132 }
133 }
134 for message in batch.messages()? {
135 let body = message.body().clone();
136 let outcome = if jobs {
137 match serde_json::from_value::<Job>(body) {
138 Ok(job) if paused && index::parked_key(&job).is_some() => search.park(&job).await,
139 Ok(job) => search.run_job(job).await,
140 Err(error) => {
141 worker::console_error!("search: a job could not be read: {error}");
142 Ok(())
143 }
144 }
145 } else {
146 match serde_json::from_value::<Event>(body) {
147 Ok(event) => search.on_event(&event).await,
148 Err(_) => Ok(()),
149 }
150 };
151 match outcome {
152 Ok(()) => message.ack(),
153 Err(error) => {
154 worker::console_error!("search: a message failed and will be retried: {error}");
155 message.retry();
156 }
157 }
158 }
159 Ok(())
160}