Sign in

flagon-io/g1t

public

g1t/services/packages/src/lib.rs

825 lines36,088 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member1//! The packages service: the registries a workspace publishes to and
2//! installs from, beside its code (docs/PACKAGES.md). Container images
3//! first, over OCI Distribution 1.1 on `g1t.sh/v2/` (oci.rs).
4//!
5//! Other services reach it over `POST /rpc/<method>`; see
6//! `g1t_contracts::packages` for the methods and their arguments. Any
7//! other request is a registry's own protocol. Files are kept through the
8//! `BlobStore` port (store/), metadata in D1 (db.rs).
9
10mod access;
Merge branch 'worktree-agent-a6a121745e81f639f'11mod archive;
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version12mod cargo;
13mod cargo_http;
Composer from the workspace's own repositories, and go get from g1t.sh14mod composer;
15mod composer_http;
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member16mod db;
17mod digest;
18mod limits;
19mod manifest;
Merge branch 'worktree-agent-ac1de8a731938ed81'20mod marshal;
Merge branch 'worktree-agent-a6a121745e81f639f'21mod maven;
22mod maven_http;
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member23mod names;
npm on g1t.sh: publish and install @<workspace>/<name> with the npm CLI and a g1t token24mod npm;
25mod npm_http;
Merge branch 'worktree-agent-a6a121745e81f639f'26mod nuget;
27mod nuget_http;
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member28mod oci;
29mod quota;
30mod range;
Merge branch 'worktree-agent-a6a121745e81f639f'31mod rubygems;
32mod rubygems_http;
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member33mod store;
34mod token;
35mod upload;
Merge branch 'worktree-agent-a6a121745e81f639f'36mod xml;
37mod yaml;
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member38
39use std::cell::RefCell;
40use std::collections::HashMap;
41
42use g1t_contracts::audit::{AuditActor, AuditTarget, NewAuditEntry, RecordAuditArgs, Surface};
43use g1t_contracts::credentials::Decision;
44use g1t_contracts::events::{Event, NewEvent, PackageEvent, Publish};
45use g1t_contracts::identity::GitCredentialsArgs;
46use g1t_contracts::packages::*;
47use g1t_contracts::repos::{GetArgs, Repo, RepoPath};
48use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
49use g1t_kit::{args, now_ms, reply, rpc_method};
50use worker::{Context, Env, Fetcher, MessageBatch, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
51
52use access::{Action, LinkedTo, Target};
53use db::{Db, PackageRow};
54use store::{BlobStore, Store};
55
56pub(crate) const SOURCE: &str = "packages";
57/// How large a request body may be: Cloudflare's limit on the zone's plan.
58const DEFAULT_MAX_REQUEST_BYTES: u64 = 100_000_000;
59/// How many packages a listing shows.
60const LIST_LIMIT: u32 = 200;
61const VERSIONS_SHOWN: u32 = 200;
62/// How much one sweep lets go of.
63const SWEEP_BATCH: u32 = 200;
64
65thread_local! {
Merge branch 'worktree-agent-ac1de8a731938ed81'66 /// Pulls counted since the last write, by package (and by version,
67 /// where it is counted too): written at most every few seconds, so a
68 /// busy image costs one write, not one a pull. What an isolate holds
69 /// when it goes away is lost: the count is approximate.
70 static DOWNLOADS: RefCell<(HashMap<DownloadKey, u64>, u64)> = RefCell::new((HashMap::new(), 0));
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member71}
72const DOWNLOADS_FLUSH_MS: u64 = 10_000;
Merge branch 'worktree-agent-ac1de8a731938ed81'73/// A package's id, and a version's when the download counts for it too.
74type DownloadKey = (String, Option<String>);
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member75
76thread_local! {
77 /// What billing allows each workspace, as asked last, and when.
78 static ALLOWANCES: RefCell<HashMap<String, (quota::Allowance, u64)>> = RefCell::new(HashMap::new());
79}
80/// How long billing's answer is kept.
81const ALLOWANCE_TTL_MS: u64 = 5 * 60 * 1000;
82
83/// Who made a registry request, as its audit entries and versions name them.
84pub struct Caller {
85 pub actor: Option<AuditActor>,
86}
87
88/// What access decisions need about a package, owned.
89pub struct TargetOf {
90 workspace: String,
91 repo: Option<(String, String, bool)>,
92 public: bool,
93}
94
95impl TargetOf {
96 pub fn package(row: &PackageRow) -> TargetOf {
97 TargetOf {
98 workspace: row.workspace.clone(),
99 repo: row
100 .repo_id
101 .as_ref()
102 .map(|id| (id.clone(), row.repo_name.clone().unwrap_or_default(), !row.public())),
103 public: row.public(),
104 }
105 }
106
107 pub fn view(&self) -> Target<'_> {
108 Target {
109 workspace: &self.workspace,
110 repo: self.repo.as_ref().map(|(id, name, private)| LinkedTo { id, name, private: *private }),
111 public: self.public,
112 }
113 }
114}
115
116pub struct Packages {
117 pub db: Db,
118 pub store: Store,
119 identity: Fetcher,
120 repos: Fetcher,
121 events: Fetcher,
122 /// Signs registry tokens (PACKAGES_TOKEN_SECRET).
123 secret: Vec<u8>,
124 max_request: u64,
125 /// The host package addresses start with: `g1t.sh`.
126 host: String,
127 /// Whether free workspaces' package storage is limited (STORAGE_LIMITS
128 /// `on`); off when self-hosted.
129 storage_limits: bool,
130 env: Env,
131}
132
133/// What a workspace's deletion or restoring does to its packages.
134#[derive(Debug, PartialEq, Eq)]
135pub(crate) enum WorkspaceMark {
136 /// Hide them from `at`, while the workspace may still be restored.
137 Hide { workspace: String, at: String },
138 /// Show them again.
139 Show { workspace: String },
140}
141
142/// The mark `event` asks for, if any. A protected workspace (`protected`,
143/// slugs or ids, as identity's PROTECTED_WORKSPACES) is never hidden: it
144/// cannot be deleted, so an event saying so is a mistake.
145pub(crate) fn workspace_mark(event: &Event, protected: &[String], now: &str) -> Option<WorkspaceMark> {
146 let text = |key: &str| event.data[key].as_str().map(|v| v.trim().to_lowercase()).filter(|v| !v.is_empty());
147 let workspace = text("slug")?;
148 match event.kind.as_str() {
149 "workspace.deleting" => {
150 let id = text("workspaceId").unwrap_or_default();
151 if protected.iter().any(|name| *name == workspace || (!id.is_empty() && *name == id)) {
152 return None;
153 }
154 Some(WorkspaceMark::Hide { workspace, at: now.to_owned() })
155 }
156 "workspace.restored" => Some(WorkspaceMark::Show { workspace }),
157 _ => None,
158 }
159}
160
161fn not_found<T>() -> Outcome<T> {
162 Outcome::fail(FailureCode::NotFound, "Package not found.")
163}
164
165impl Packages {
166 pub fn from_env(env: &Env) -> Result<Packages> {
167 let secret = store::var(env, "PACKAGES_TOKEN_SECRET");
168 if secret.len() < 32 {
169 return Err(worker::Error::RustError("PACKAGES_TOKEN_SECRET is missing or shorter than 32 characters".into()));
170 }
171 // 0 is no limit, as self-hosted (nothing in front cuts bodies short).
172 let max_request = match store::var(env, "MAX_REQUEST_BYTES").parse::<u64>() {
173 Ok(0) => u64::MAX,
174 Ok(limit) => limit,
175 Err(_) => DEFAULT_MAX_REQUEST_BYTES,
176 };
177 let host = store::var(env, "REGISTRY_HOST");
178 Ok(Packages {
179 db: Db { db: env.d1("DB")? },
Merge branch 'worktree-agent-ac5b181a013e54348'180 store: store::from_env(env)?,
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member181 identity: env.service("IDENTITY")?,
182 repos: env.service("REPOS")?,
183 events: env.service("EVENTS")?,
184 secret: secret.into_bytes(),
185 max_request,
186 host: if host.is_empty() { "g1t.sh".to_owned() } else { host },
187 storage_limits: store::var(env, "STORAGE_LIMITS") == "on",
188 env: env.clone(),
189 })
190 }
191
192 async fn viewer_for(&self, username: &str, secret: &str) -> Result<Viewer> {
193 g1t_kit::call(
194 &self.identity,
195 "user_for_git_credentials",
196 &GitCredentialsArgs { username: username.to_owned(), secret: secret.to_owned() },
197 )
198 .await
199 }
200
201 /// A repository of the workspace by name, whoever may see it.
202 async fn repo_by_name(&self, workspace: &str, name: &str) -> Result<Option<Repo>> {
203 if !g1t_contracts::is_valid_repo_name(name) {
204 return Ok(None);
205 }
206 let found: Outcome<Repo> = g1t_kit::call(
207 &self.repos,
208 "get",
209 &GetArgs {
210 path: RepoPath { namespace: workspace.to_owned(), name: name.to_owned() },
211 viewer: Some(User::system(workspace)),
212 },
213 )
214 .await?;
215 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()))
216 }
217
218 /// Who may do what to an image: the package's own settings, or, before
219 /// its first push, the repository it will be linked to.
220 async fn target(&self, name: &names::ImageName, found: Option<&PackageRow>) -> Result<TargetOf> {
221 if let Some(found) = found {
222 return Ok(TargetOf::package(found));
223 }
224 let repo = self.repo_by_name(&name.workspace, name.repo_name()).await?;
225 Ok(TargetOf {
226 workspace: name.workspace.clone(),
227 repo: repo.map(|repo| (repo.id, repo.name, repo.is_private)),
228 public: false,
229 })
230 }
231
232 /// What billing allows the workspace, kept for five minutes. `None`
233 /// when billing cannot be asked: the push is then let through.
234 async fn allowance(&self, workspace: &str) -> Option<quota::Allowance> {
235 let now = now_ms();
236 let kept = ALLOWANCES.with(|kept| kept.borrow().get(workspace).copied());
237 if let Some((allowance, at)) = kept
238 && now.saturating_sub(at) < ALLOWANCE_TTL_MS
239 {
240 return Some(allowance);
241 }
242 let billing = self.env.service("BILLING").ok()?;
243 let asked: Result<quota::Allowance> = g1t_kit::call(
244 &billing,
245 "entitlements",
246 &g1t_contracts::billing::EntitlementsArgs { workspace: workspace.to_owned() },
247 )
248 .await;
249 match asked {
250 Ok(allowance) => {
251 ALLOWANCES.with(|kept| kept.borrow_mut().insert(workspace.to_owned(), (allowance, now)));
252 Some(allowance)
253 }
254 Err(error) => {
255 worker::console_error!("packages: billing could not be asked about {workspace}, letting the push through: {error}");
256 None
257 }
258 }
259 }
260
261 /// Why a push of these files (digest and size) into the package may
262 /// not be kept, if it may not: a free workspace past its free storage.
263 /// Files the workspace already holds add nothing.
264 pub(crate) async fn storage_refusal(&self, package: &PackageRow, files: &[(String, u64)]) -> Result<Option<String>> {
265 if !self.storage_limits || files.is_empty() {
266 return Ok(None);
267 }
268 let digests: Vec<String> = files.iter().map(|(digest, _)| digest.clone()).collect();
269 let held = self.db.held(&package.workspace, &digests).await?;
270 let mut seen = std::collections::HashSet::new();
271 let adding: u64 = files
272 .iter()
273 .filter(|(digest, _)| !held.contains(digest) && seen.insert(digest.clone()))
274 .map(|(_, size)| size)
275 .sum();
276 if adding == 0 {
277 return Ok(None);
278 }
279 let Some(allowance) = self.allowance(&package.workspace).await else {
280 return Ok(None);
281 };
282 let (public_bytes, private_bytes) = self.db.storage(&package.workspace).await?;
283 let public = package.public();
284 let used = if public { public_bytes } else { private_bytes };
285 Ok(quota::decide(&allowance, public, used, adding)
286 .err()
287 .map(|refusal| quota::message(&package.workspace, &refusal)))
288 }
289
290 fn count_download(&self, package_id: &str, ctx: &Context) {
Merge branch 'worktree-agent-ac1de8a731938ed81'291 self.count_downloads(package_id, None, ctx);
292 }
293
294 /// A download of one version, counted for it and its package.
295 fn count_version_download(&self, package_id: &str, version_id: &str, ctx: &Context) {
296 self.count_downloads(package_id, Some(version_id), ctx);
297 }
298
299 fn count_downloads(&self, package_id: &str, version_id: Option<&str>, ctx: &Context) {
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member300 let due = DOWNLOADS.with(|counts| {
301 let mut counts = counts.borrow_mut();
Merge branch 'worktree-agent-ac1de8a731938ed81'302 *counts.0.entry((package_id.to_owned(), version_id.map(str::to_owned))).or_default() += 1;
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member303 let now = now_ms();
304 if now.saturating_sub(counts.1) < DOWNLOADS_FLUSH_MS {
305 return None;
306 }
307 counts.1 = now;
308 Some(counts.0.drain().collect::<Vec<_>>())
309 });
310 if let Some(due) = due {
311 let env = self.env.clone();
312 ctx.wait_until(async move {
313 let written = match env.d1("DB") {
314 Ok(db) => Db { db }.add_downloads(&due).await,
315 Err(error) => Err(error),
316 };
317 if let Err(error) = written {
318 worker::console_error!("packages: downloads not counted: {error}");
319 }
320 });
321 }
322 }
323
324 async fn announce(&self, kind: &'static str, package: &PackageRow, data: PackageEvent, caller: &Caller) {
325 let event = NewEvent {
326 kind,
327 source: SOURCE,
328 repo_id: package.repo_id.clone(),
329 actor: caller.actor.as_ref().map(|actor| actor.actor_id.clone()),
330 data,
331 };
332 let published: Result<serde_json::Value> = g1t_kit::call(&self.events, "publish", &Publish { events: vec![event] }).await;
333 if let Err(error) = published {
334 worker::console_error!("packages: {kind} not published: {error}");
335 }
336 }
337
338 async fn audit(&self, caller: &Caller, action: &str, package: &PackageRow, path: Option<&str>, surface: Option<Surface>) {
339 let Some(actor) = caller.actor.clone() else {
340 return;
341 };
342 let target = AuditTarget {
343 workspace: package.workspace.clone(),
344 repo: package.repo_name.as_ref().map(|name| format!("{}/{name}", package.workspace)),
345 path: Some(path.map_or_else(|| format!("{}:{}/{}", package.ecosystem, package.workspace, package.name), str::to_owned)),
346 ..AuditTarget::default()
347 };
348 let mut entry = NewAuditEntry::new(
349 actor,
350 action,
351 surface.unwrap_or(Surface::Registry),
352 target,
353 &Decision::allow("packages"),
354 new_id("req", now_ms()),
355 );
356 entry.result = Some("ok".to_owned());
357 let recorded: Result<u32> = g1t_kit::call(&self.events, "audit_record", &RecordAuditArgs { entries: vec![entry] }).await;
358 if let Err(error) = recorded {
359 worker::console_error!("packages: audit entry not recorded: {error}");
360 }
361 }
362
363 fn summary(&self, row: &db::ListedRow) -> PackageSummary {
364 let p = &row.package;
365 PackageSummary {
366 id: p.id.clone(),
367 workspace: p.workspace.clone(),
368 ecosystem: Ecosystem::parse(&p.ecosystem).unwrap_or(Ecosystem::Container),
369 name: p.name.clone(),
npm on g1t.sh: publish and install @<workspace>/<name> with the npm CLI and a g1t token370 address: match p.ecosystem.as_str() {
371 "npm" => format!("{}/-/npm/@{}/{}", self.host, p.workspace, p.name),
Composer from the workspace's own repositories, and go get from g1t.sh372 "composer" => format!("{}/-/composer/{}/{}", self.host, p.workspace, p.name),
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version373 "cargo" => format!("{}/-/cargo/{}/{}", self.host, p.workspace, p.name),
Merge branch 'worktree-agent-a6a121745e81f639f'374 "maven" => format!("{}/-/maven/{}/{}", self.host, p.workspace, p.name),
375 "nuget" => format!("{}/-/nuget/{}/{}", self.host, p.workspace, p.name),
376 "rubygems" => format!("{}/-/rubygems/{}/{}", self.host, p.workspace, p.name),
npm on g1t.sh: publish and install @<workspace>/<name> with the npm CLI and a g1t token377 _ => format!("{}/{}/{}", self.host, p.workspace, p.name),
378 },
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member379 visibility: Visibility::parse(&p.visibility),
380 repo: p.repo_id.as_ref().map(|id| LinkedRepo {
381 id: id.clone(),
382 namespace: p.workspace.clone(),
383 name: p.repo_name.clone().unwrap_or_default(),
384 }),
385 description: p.description.clone(),
386 versions: row.version_count,
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version387 latest: db::latest_shown(row),
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member388 size: row.bytes,
389 downloads: p.downloads,
390 created_at: p.created_at.clone(),
391 updated_at: p.updated_at.clone(),
392 }
393 }
394
395 async fn listed(&self, workspace: &str, ecosystem: Ecosystem, name: &str) -> Result<Option<db::ListedRow>> {
396 let rows = self.db.list(workspace, Some(ecosystem.as_str()), None, None, LIST_LIMIT).await?;
397 if let Some(row) = rows.into_iter().find(|row| row.package.name == name) {
398 return Ok(Some(row));
399 }
400 // Past the first page: read it alone.
401 Ok(self.db.package(workspace, ecosystem.as_str(), name).await?.filter(|package| !package.hidden()).map(|package| db::ListedRow {
402 package,
403 version_count: 0,
404 bytes: 0,
405 latest_tag: None,
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version406 latest_tag_version: None,
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member407 latest_version: None,
408 }))
409 }
410
411 async fn list_packages(&self, a: ListPackagesArgs) -> Result<Outcome<Vec<PackageSummary>>> {
412 let workspace = a.workspace.to_lowercase();
413 let rows = self
414 .db
415 .list(&workspace, a.ecosystem.map(Ecosystem::as_str), a.repo_id.as_deref(), a.query.as_deref(), LIST_LIMIT)
416 .await?;
417 let visible = rows
418 .iter()
419 .filter(|row| access::decide(a.viewer.as_ref(), &TargetOf::package(&row.package).view(), Action::Pull).allowed)
420 .map(|row| self.summary(row))
421 .collect();
422 Ok(Outcome::Ok(visible))
423 }
424
425 async fn get_package(&self, a: GetPackageArgs) -> Result<Outcome<PackageDetail>> {
426 let workspace = a.workspace.to_lowercase();
427 let Some(row) = self.listed(&workspace, a.ecosystem, &a.name).await? else {
428 return Ok(not_found());
429 };
430 let target = TargetOf::package(&row.package);
431 let permissions = access::permissions(a.viewer.as_ref(), &target.view());
432 if !permissions.pull {
433 return Ok(not_found());
434 }
435 let tags = self.db.tags(&row.package.id).await?;
Merge branch 'worktree-agent-a6a121745e81f639f'436 // A yanked or unlisted version reads as deprecated: still there
437 // for lockfiles that name it, no longer picked for new ones.
438 let withdrawn = match row.package.ecosystem.as_str() {
439 "nuget" => "Unlisted: still restored by projects that name it, no longer shown in search.",
440 "rubygems" => "Yanked: Bundler no longer picks this version for new lockfiles.",
441 _ => "Yanked: Cargo no longer picks this version for new lockfiles.",
442 };
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member443 let versions = self
444 .db
445 .versions(&row.package.id, VERSIONS_SHOWN)
446 .await?
447 .into_iter()
448 .map(|version| {
449 let meta = version.meta();
450 let text = |key: &str| meta[key].as_str().map(str::to_owned);
451 PackageVersion {
452 tags: tags.iter().filter(|tag| tag.version_id == version.id).map(|tag| tag.tag.clone()).collect(),
453 media_type: text("media_type"),
454 artifact_type: text("artifact_type"),
455 platforms: meta["platforms"]
456 .as_array()
457 .map(|list| list.iter().filter_map(|p| p.as_str().map(str::to_owned)).collect())
458 .unwrap_or_default(),
459 id: version.id,
460 version: version.version,
461 digest: version.digest,
462 size: version.size,
463 subject: version.subject,
464 published_by: version.published_by,
465 published_at: version.published_at,
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version466 deprecated: if version.yanked != 0 {
Merge branch 'worktree-agent-a6a121745e81f639f'467 Some(withdrawn.to_owned())
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version468 } else {
469 version.deprecated
470 },
Merge branch 'worktree-agent-ac1de8a731938ed81'471 symbols: meta["symbols"] == true,
472 downloads: (row.package.ecosystem == "nuget").then_some(version.downloads),
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member473 }
474 })
475 .collect();
npm on g1t.sh: publish and install @<workspace>/<name> with the npm CLI and a g1t token476 // The README its page shows: npm's, from the latest version.
477 let readme = match self.db.readme_digest(&row.package.id).await?.and_then(|d| digest::Digest::parse(&d)) {
478 Some(digest) => match self.db.blob(&digest).await? {
479 Some(blob) => self.store.read(&blob.object_key).await?.map(|b| String::from_utf8_lossy(&b).into_owned()),
480 None => None,
481 },
482 None => None,
483 };
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member484 Ok(Outcome::Ok(PackageDetail {
npm on g1t.sh: publish and install @<workspace>/<name> with the npm CLI and a g1t token485 readme,
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member486 package: self.summary(&row),
487 versions,
488 tags: tags
489 .into_iter()
490 .map(|tag| PackageTag { tag: tag.tag, digest: tag.digest, updated_at: tag.updated_at })
491 .collect(),
492 permissions,
493 }))
494 }
495
496 /// The package `actor` asks to change, if they may `action` it.
497 async fn for_change(&self, actor: &User, workspace: &str, ecosystem: Ecosystem, name: &str, action: Action) -> Result<Outcome<PackageRow>> {
498 let Some(package) = self.db.package(&workspace.to_lowercase(), ecosystem.as_str(), name).await?.filter(|p| !p.hidden()) else {
499 return Ok(not_found());
500 };
501 let target = TargetOf::package(&package);
502 let decision = access::decide(Some(actor), &target.view(), action);
503 if decision.allowed {
504 return Ok(Outcome::Ok(package));
505 }
506 if !access::decide(Some(actor), &target.view(), Action::Pull).allowed {
507 return Ok(not_found());
508 }
509 Ok(Outcome::fail(FailureCode::Forbidden, decision.reason.unwrap_or_else(|| "Not allowed.".to_owned())))
510 }
511
512 async fn delete_version(&self, a: DeleteVersionArgs) -> Result<Outcome<()>> {
513 let package = match self.for_change(&a.actor, &a.workspace, a.ecosystem, &a.name, Action::Delete).await? {
514 Outcome::Ok(package) => package,
515 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
516 };
517 let Some(version) = self.db.find_version(&package.id, &a.version).await? else {
518 return Ok(Outcome::fail(FailureCode::NotFound, "Version not found."));
519 };
520 let caller = Caller { actor: Some(AuditActor::of(&a.actor)) };
521 self.remove_version(&package, &version, &caller).await?;
522 Ok(Outcome::Ok(()))
523 }
524
525 async fn delete_package(&self, a: DeletePackageArgs) -> Result<Outcome<()>> {
526 let package = match self.for_change(&a.actor, &a.workspace, a.ecosystem, &a.name, Action::Delete).await? {
527 Outcome::Ok(package) => package,
528 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
529 };
530 self.db.delete_package(&package.id).await?;
531 self.db.measure(&package.workspace).await?;
532 let caller = Caller { actor: Some(AuditActor::of(&a.actor)) };
533 self.announce("package.deleted", &package, self.event_of(&package), &caller).await;
534 self.audit(&caller, "package.delete", &package, None, a.surface).await;
535 Ok(Outcome::Ok(()))
536 }
537
538 async fn set_package(&self, a: SetPackageArgs) -> Result<Outcome<PackageSummary>> {
539 let package = match self.for_change(&a.actor, &a.workspace, a.ecosystem, &a.name, Action::Delete).await? {
540 Outcome::Ok(package) => package,
541 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
542 };
543 let caller = Caller { actor: Some(AuditActor::of(&a.actor)) };
544 let now = now_ms();
545 let before = package.visibility.clone();
546 if a.unlink {
547 self.db.set_link(&package.id, None, "private", now).await?;
548 } else if let Some(link) = a.link.as_deref() {
549 let Some(repo) = self.repo_by_name(&package.workspace, link).await? else {
550 return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no repository {}/{link}.", package.workspace)));
551 };
552 // Linking hands the package to the repository's roles: only
553 // someone who administers that repository may.
554 let role = g1t_contracts::access::permission(Some(&a.actor), &repo);
555 if role < Some(g1t_contracts::access::RepoRole::Admin) {
556 return Ok(Outcome::fail(
557 FailureCode::Forbidden,
558 format!("You need the Admin role on {}/{} to link a package to it.", repo.namespace, repo.name),
559 ));
560 }
561 let visibility = if repo.is_private { "private" } else { "public" };
562 self.db.set_link(&package.id, Some((&repo.id, &repo.name)), visibility, now).await?;
563 }
564 if let Some(visibility) = a.visibility {
565 let linked = if a.unlink { false } else { a.link.is_some() || package.repo_id.is_some() };
566 if linked {
567 return Ok(Outcome::fail(
568 FailureCode::Invalid,
569 "A linked package has its repository's visibility. Change the repository's, or unlink the package.",
570 ));
571 }
572 self.db.set_visibility(&package.id, visibility.as_str(), now).await?;
573 }
574 self.audit(&caller, "package.update", &package, None, a.surface).await;
575 let Some(row) = self.listed(&package.workspace, a.ecosystem, &package.name).await? else {
576 return Ok(not_found());
577 };
578 if row.package.visibility != before {
579 self.db.measure(&package.workspace).await?;
580 let event = PackageEvent { visibility: Some(row.package.visibility.clone()), ..self.event_of(&row.package) };
581 self.announce("package.visibility_changed", &row.package, event, &caller).await;
582 }
583 Ok(Outcome::Ok(self.summary(&row)))
584 }
585
586 /// Lets go of expired uploads, and of blobs no version has used for a day.
587 async fn sweep(&self) -> Result<(u32, u32)> {
588 let now = now_ms();
589 let mut uploads = 0;
590 for row in self.db.expired_uploads(now, SWEEP_BATCH).await? {
591 if let Some(progress) = row.progress()
592 && let Err(error) = upload::abort(&self.store, &progress).await
593 {
594 worker::console_error!("packages: upload {} not aborted: {error}", row.id);
595 }
596 self.db.delete_upload(&row.id).await?;
597 uploads += 1;
598 }
599 let mut blobs = 0;
600 for blob in self.db.unused_blobs(now, SWEEP_BATCH).await? {
601 self.store.delete(&blob.object_key).await?;
602 self.db.forget_blob(&blob.digest).await?;
603 blobs += 1;
604 }
605 Ok((uploads, blobs))
606 }
607
608 /// Follows what happens elsewhere: workspaces renamed and deleted,
609 /// repositories that change visibility, are renamed, move or go.
610 async fn on_event(&self, env: &Env, event: &Event) -> Result<()> {
Composer from the workspace's own repositories, and go get from g1t.sh611 // Composer packages are read from repositories: again when one is
612 // pushed to (its default branch may have gained a composer.json),
613 // restored, renamed or moved; gone when it is deleted. A failure is
614 // logged, not retried with the batch: the next push reads it again.
615 let repo_id = event.repo_id.clone().or_else(|| event.data["repoId"].as_str().map(str::to_owned));
616 if let Some(repo_id) = repo_id.as_deref() {
617 let synced = match event.kind.as_str() {
618 "git.push" => {
619 let known = self.db.package_for_repo(repo_id, "composer").await?.is_some();
620 if known || event.data["defaultBranch"].as_bool() == Some(true) {
621 Some(self.sync_composer(repo_id).await.map(|_| ()))
622 } else {
623 None
624 }
625 }
626 "repo.deleted" => Some(self.composer_repo_gone(repo_id).await),
627 "repo.restored" | "repo.renamed" | "repo.transferred" => Some(self.sync_composer(repo_id).await.map(|_| ())),
628 _ => None,
629 };
630 if let Some(Err(error)) = synced {
631 worker::console_error!("packages: composer {} for {repo_id}: {error}", event.kind);
632 }
633 if event.kind == "git.push" || event.kind == "repo.deleted" || event.kind == "repo.restored" {
634 return Ok(());
635 }
636 }
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member637 let db = &self.db.db;
638 let protected = g1t_contracts::identity::protected_names(Some(&store::var(env, "PROTECTED_WORKSPACES")));
639 match workspace_mark(event, &protected, &g1t_contracts::time::rfc3339(now_ms())) {
640 Some(WorkspaceMark::Hide { workspace, at }) => return self.db.mark_workspace(&workspace, Some(&at)).await,
641 Some(WorkspaceMark::Show { workspace }) => return self.db.mark_workspace(&workspace, None).await,
642 None => {}
643 }
644 if g1t_kit::rename::on_event(env, db, event, &[
645 "UPDATE packages SET workspace = ?1 WHERE workspace = ?2",
646 "UPDATE uploads SET workspace = ?1 WHERE workspace = ?2",
647 "UPDATE workspace_blobs SET workspace = ?1 WHERE workspace = ?2",
648 ])
649 .await?
650 {
651 return Ok(());
652 }
653 if g1t_kit::deleted::on_event(db, event, &[
654 "DELETE FROM tags WHERE package_id IN (SELECT id FROM packages WHERE workspace = ?1)",
655 "DELETE FROM version_files WHERE version_id IN (SELECT v.id FROM versions v JOIN packages p ON p.id = v.package_id WHERE p.workspace = ?1)",
656 "DELETE FROM versions WHERE package_id IN (SELECT id FROM packages WHERE workspace = ?1)",
657 "DELETE FROM package_blobs WHERE package_id IN (SELECT id FROM packages WHERE workspace = ?1)",
658 "DELETE FROM uploads WHERE workspace = ?1",
659 "DELETE FROM workspace_blobs WHERE workspace = ?1",
660 "DELETE FROM packages WHERE workspace = ?1",
661 ])
662 .await?
663 {
664 return Ok(());
665 }
666 // A repository renamed keeps its packages linked under its new
667 // name; one moved to another workspace leaves them behind, unlinked.
668 if g1t_kit::transfer::on_event(env, db, event, &[
669 "UPDATE packages SET repo_name = ?6 WHERE repo_id = ?5 AND workspace = ?3",
670 "UPDATE packages SET repo_id = NULL, repo_name = NULL WHERE repo_id = ?5 AND workspace <> ?3",
671 ])
672 .await?
673 {
674 return Ok(());
675 }
676 if g1t_kit::lifecycle::on_purged(db, event, &["UPDATE packages SET repo_id = NULL, repo_name = NULL, visibility = 'private' WHERE repo_id = ?1"])
677 .await?
678 {
679 return Ok(());
680 }
681 if event.kind == "repo.visibility_changed" {
682 #[derive(serde::Deserialize)]
683 #[serde(rename_all = "camelCase")]
684 struct Changed {
685 repo_id: String,
686 is_private: bool,
687 }
688
689 let Ok(changed) = serde_json::from_value::<Changed>(event.data.clone()) else {
690 worker::console_error!("repo.visibility_changed {} could not be read", event.id);
691 return Ok(());
692 };
693 self.db.follow_visibility(&changed.repo_id, changed.is_private).await?;
694 for workspace in self.db.workspaces_linked_to(&changed.repo_id).await? {
695 self.db.measure(&workspace).await?;
696 }
697 }
698 Ok(())
699 }
700}
701
702#[event(fetch)]
703async fn fetch(mut request: Request, env: Env, ctx: Context) -> Result<Response> {
704 let packages = Packages::from_env(&env)?;
705 let Some(method) = rpc_method(&request) else {
npm on g1t.sh: publish and install @<workspace>/<name> with the npm CLI and a g1t token706 if request.path().starts_with("/-/npm/") || request.path() == "/-/npm" {
707 return packages.npm(request, &ctx).await;
708 }
Composer from the workspace's own repositories, and go get from g1t.sh709 if request.path().starts_with("/-/composer/") {
710 return packages.composer(request, &ctx).await;
711 }
Cargo on g1t.sh: a sparse registry per workspace, cargo publish/add/yank/search with a g1t token; npm rows show their version712 if request.path().starts_with("/-/cargo/") {
713 return packages.cargo(request, &ctx).await;
714 }
Merge branch 'worktree-agent-a6a121745e81f639f'715 if request.path().starts_with("/-/maven/") {
716 return packages.maven(request, &ctx).await;
717 }
718 if request.path().starts_with("/-/nuget/") {
719 return packages.nuget(request, &ctx).await;
720 }
721 if request.path().starts_with("/-/rubygems/") {
722 return packages.rubygems(request, &ctx).await;
723 }
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member724 return packages.registry(request, &ctx).await;
725 };
726 let body: serde_json::Value = request.json().await?;
727 match method.as_str() {
728 "list_packages" => reply(&packages.list_packages(args(body)?).await?),
729 "get_package" => reply(&packages.get_package(args(body)?).await?),
730 "delete_version" => reply(&packages.delete_version(args(body)?).await?),
731 "delete_package" => reply(&packages.delete_package(args(body)?).await?),
732 "set_package" => reply(&packages.set_package(args(body)?).await?),
733 // For billing: what a workspace's packages hold.
734 "storage_all" => reply(&packages.db.storage_all().await?),
Composer from the workspace's own repositories, and go get from g1t.sh735 // Read a repository's Composer package again now, as a push would.
736 "sync_composer" => {
737 let a: SyncComposerArgs = args(body)?;
738 reply(&packages.sync_composer(&a.repo_id).await?)
739 }
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member740 "storage" => {
741 let a: StorageArgs = args(body)?;
742 let (public_bytes, private_bytes) = packages.db.storage(&a.workspace.to_lowercase()).await?;
743 reply(&PackageStorage { public_bytes, private_bytes })
744 }
745 _ => Response::error("Unknown method", 404),
746 }
747}
748
749/// Every hour: expired uploads are let go, and blobs no version has used
750/// for a day are deleted from the store.
751#[event(scheduled)]
752async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
753 let packages = match Packages::from_env(&env) {
754 Ok(packages) => packages,
755 Err(error) => {
756 worker::console_error!("packages: the sweep could not start: {error}");
757 return;
758 }
759 };
760 match packages.sweep().await {
761 Ok((0, 0)) => {}
762 Ok((uploads, blobs)) => worker::console_log!("packages: let go of {uploads} uploads and {blobs} blobs"),
763 Err(error) => worker::console_error!("packages: the sweep failed: {error}"),
764 }
Composer from the workspace's own repositories, and go get from g1t.sh765 // Repositories that had a composer.json before the registry did.
766 match packages.composer_backfill().await {
767 Ok(0) => {}
768 Ok(found) => worker::console_log!("packages: found {found} Composer packages"),
769 Err(error) => worker::console_error!("packages: the Composer backfill failed: {error}"),
770 }
Packages, with a container registry on g1t.sh; workspaces deleted whole and kept 30 days; Members for every member771}
772
773#[event(queue)]
774async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
775 let packages = Packages::from_env(&env)?;
776 for message in batch.messages()? {
777 packages.on_event(&env, message.body()).await?;
778 }
779 Ok(())
780}
781
782#[cfg(test)]
783mod tests {
784 use super::*;
785 use serde_json::json;
786
787 fn event(kind: &str, data: serde_json::Value) -> Event {
788 Event {
789 id: "evt_1".into(),
790 kind: kind.into(),
791 source: "identity".into(),
792 time: "2026-10-06T00:00:00.000Z".into(),
793 repo_id: None,
794 actor: None,
795 data,
796 }
797 }
798
799 #[test]
800 fn a_deleted_workspace_is_hidden_and_a_restored_one_shown() {
801 let protected = g1t_contracts::identity::protected_names(Some("wsp_keep"));
802 let now = "2026-10-06T12:00:00.000Z";
803 let deleting = event("workspace.deleting", json!({ "workspaceId": "wsp_1", "slug": "Acme", "by": "ana", "purgeAfter": "x" }));
804 assert_eq!(
805 workspace_mark(&deleting, &protected, now),
806 Some(WorkspaceMark::Hide { workspace: "acme".into(), at: now.into() })
807 );
808 let restored = event("workspace.restored", json!({ "workspaceId": "wsp_1", "slug": "acme" }));
809 assert_eq!(workspace_mark(&restored, &protected, now), Some(WorkspaceMark::Show { workspace: "acme".into() }));
810 }
811
812 #[test]
813 fn protected_workspaces_and_other_events_are_left_alone() {
814 let protected = g1t_contracts::identity::protected_names(Some("wsp_keep"));
815 let now = "2026-10-06T12:00:00.000Z";
816 let flagon = event("workspace.deleting", json!({ "workspaceId": "wsp_f", "slug": "flagon-io" }));
817 assert_eq!(workspace_mark(&flagon, &protected, now), None, "always protected");
818 let by_id = event("workspace.deleting", json!({ "workspaceId": "wsp_keep", "slug": "kept" }));
819 assert_eq!(workspace_mark(&by_id, &protected, now), None, "protected by id");
820 let purged = event("workspace.deleted", json!({ "workspaceId": "wsp_1", "slug": "acme" }));
821 assert_eq!(workspace_mark(&purged, &protected, now), None, "the purge is handled apart");
822 let nameless = event("workspace.deleting", json!({ "workspaceId": "wsp_1", "slug": "" }));
823 assert_eq!(workspace_mark(&nameless, &protected, now), None);
824 }
825}

This file's history is long; its oldest lines are credited to the oldest commit read.