package database import ( "database/sql" "errors" "fmt" "sort" "time" "github.com/git-pkgs/artifacts" "github.com/git-pkgs/purl" "github.com/opencontainers/go-digest" ) // Package queries func (db *DB) GetPackageByPURL(purl string) (*Package, error) { var pkg Package query := db.Rebind(` SELECT id, purl, ecosystem, name, latest_version, license, description, homepage, repository_url, registry_url, supplier_name, supplier_type, source, enriched_at, vulns_synced_at, created_at, updated_at FROM packages WHERE purl = ? `) err := db.Get(&pkg, query, purl) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &pkg, nil } func (db *DB) GetPackageByEcosystemName(ecosystem, name string) (*Package, error) { var pkg Package query := db.Rebind(` SELECT id, purl, ecosystem, name, latest_version, license, description, homepage, repository_url, registry_url, supplier_name, supplier_type, source, enriched_at, vulns_synced_at, created_at, updated_at FROM packages WHERE ecosystem = ? AND name = ? `) err := db.Get(&pkg, query, ecosystem, name) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &pkg, nil } func (db *DB) UpsertPackage(pkg *Package) error { now := time.Now() var query string if db.dialect == DialectPostgres { query = ` INSERT INTO packages (purl, ecosystem, name, latest_version, license, description, homepage, repository_url, registry_url, enriched_at, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT(purl) DO UPDATE SET latest_version = EXCLUDED.latest_version, license = EXCLUDED.license, description = EXCLUDED.description, homepage = EXCLUDED.homepage, repository_url = EXCLUDED.repository_url, registry_url = EXCLUDED.registry_url, enriched_at = EXCLUDED.enriched_at, updated_at = EXCLUDED.updated_at ` } else { query = ` INSERT INTO packages (purl, ecosystem, name, latest_version, license, description, homepage, repository_url, registry_url, enriched_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(purl) DO UPDATE SET latest_version = excluded.latest_version, license = excluded.license, description = excluded.description, homepage = excluded.homepage, repository_url = excluded.repository_url, registry_url = excluded.registry_url, enriched_at = excluded.enriched_at, updated_at = excluded.updated_at ` } _, err := db.Exec(query, pkg.PURL, pkg.Ecosystem, pkg.Name, pkg.LatestVersion, pkg.License, pkg.Description, pkg.Homepage, pkg.RepositoryURL, pkg.RegistryURL, pkg.EnrichedAt, now, now, ) if err != nil { return fmt.Errorf("upserting package: %w", err) } return nil } // Version queries func (db *DB) GetVersionByPURL(purl string) (*Version, error) { var v Version query := db.Rebind(` SELECT id, purl, package_purl, license, integrity, published_at, yanked, source, enriched_at, created_at, updated_at FROM versions WHERE purl = ? `) err := db.Get(&v, query, purl) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &v, nil } func (db *DB) GetVersionsByPackagePURL(packagePURL string) ([]Version, error) { var versions []Version query := db.Rebind(` SELECT id, purl, package_purl, license, integrity, published_at, yanked, source, enriched_at, created_at, updated_at FROM versions WHERE package_purl = ? ORDER BY created_at DESC `) err := db.Select(&versions, query, packagePURL) if err != nil { return nil, err } return versions, nil } func (db *DB) UpsertVersion(v *Version) error { now := time.Now() var query string if db.dialect == DialectPostgres { query = ` INSERT INTO versions (purl, package_purl, license, integrity, published_at, yanked, enriched_at, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT(purl) DO UPDATE SET license = EXCLUDED.license, integrity = EXCLUDED.integrity, published_at = COALESCE(EXCLUDED.published_at, versions.published_at), yanked = EXCLUDED.yanked, enriched_at = EXCLUDED.enriched_at, updated_at = EXCLUDED.updated_at ` } else { query = ` INSERT INTO versions (purl, package_purl, license, integrity, published_at, yanked, enriched_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(purl) DO UPDATE SET license = excluded.license, integrity = excluded.integrity, published_at = COALESCE(excluded.published_at, published_at), yanked = excluded.yanked, enriched_at = excluded.enriched_at, updated_at = excluded.updated_at ` } _, err := db.Exec(query, v.PURL, v.PackagePURL, v.License, v.Integrity, v.PublishedAt, v.Yanked, v.EnrichedAt, now, now, ) if err != nil { return fmt.Errorf("upserting version: %w", err) } return nil } // SetVersionPublishedAt records a version's publish time, creating the // versions row if the proxy has not seen the version yet. It only writes // published_at, so it never disturbs enrichment data on an existing row. func (db *DB) SetVersionPublishedAt(versionPURL, packagePURL string, publishedAt time.Time) error { now := time.Now() var query string if db.dialect == DialectPostgres { query = ` INSERT INTO versions (purl, package_purl, published_at, created_at, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT(purl) DO UPDATE SET published_at = EXCLUDED.published_at, updated_at = EXCLUDED.updated_at ` } else { query = ` INSERT INTO versions (purl, package_purl, published_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(purl) DO UPDATE SET published_at = excluded.published_at, updated_at = excluded.updated_at ` } _, err := db.Exec(query, versionPURL, packagePURL, publishedAt, now, now) if err != nil { return fmt.Errorf("setting version publish time: %w", err) } return nil } // Artifact queries func (db *DB) GetArtifact(versionPURL, filename string) (*Artifact, error) { var a Artifact query := db.Rebind(` SELECT id, version_purl, filename, upstream_url, storage_path, content_hash, size, content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at, source FROM artifacts WHERE version_purl = ? AND filename = ? `) err := db.Get(&a, query, versionPURL, filename) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &a, nil } // GetCachedArtifact returns the fields needed to serve a cached artifact. func (db *DB) GetCachedArtifact(packagePURL, versionPURL, filename string) (*CachedArtifact, error) { var row cachedArtifactRow query := db.Rebind(` SELECT packages.ecosystem, artifacts.storage_path, artifacts.content_hash, artifacts.size, artifacts.content_type, versions.integrity FROM artifacts JOIN versions ON versions.purl = artifacts.version_purl JOIN packages ON packages.purl = versions.package_purl WHERE packages.purl = ? AND artifacts.version_purl = ? AND artifacts.filename = ? AND artifacts.storage_path IS NOT NULL AND artifacts.fetched_at IS NOT NULL `) err := db.Get(&row, query, packagePURL, versionPURL, filename) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return row.artifact(versionPURL, filename), nil } type cachedArtifactRow struct { Ecosystem string `db:"ecosystem"` StoragePath string `db:"storage_path"` ContentHash sql.NullString `db:"content_hash"` Size sql.NullInt64 `db:"size"` ContentType sql.NullString `db:"content_type"` Integrity sql.NullString `db:"integrity"` } // artifact converts a cached artifact row to a CachedArtifact without // validation. A malformed hash or integrity value is handled by // checkCache, which clears the record and treats the request as a cache // miss so the client is served a fresh fetch instead of an error. func (row cachedArtifactRow) artifact(versionPURL, filename string) *CachedArtifact { return &CachedArtifact{ Ecosystem: row.Ecosystem, StoragePath: row.StoragePath, Integrity: row.Integrity, Artifact: artifacts.Artifact{ PURL: versionPURL, Digest: digest.Digest("sha256:" + row.ContentHash.String), Size: row.Size.Int64, Filename: filename, MediaType: row.ContentType.String, }, } } func (db *DB) GetArtifactByPath(storagePath string) (*Artifact, error) { var a Artifact query := db.Rebind(` SELECT id, version_purl, filename, upstream_url, storage_path, content_hash, size, content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at, source FROM artifacts WHERE storage_path = ? `) err := db.Get(&a, query, storagePath) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &a, nil } func (db *DB) GetArtifactsByVersionPURL(versionPURL string) ([]Artifact, error) { var artifacts []Artifact query := db.Rebind(` SELECT id, version_purl, filename, upstream_url, storage_path, content_hash, size, content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at, source FROM artifacts WHERE version_purl = ? ORDER BY filename `) err := db.Select(&artifacts, query, versionPURL) if err != nil { return nil, err } return artifacts, nil } // upsertArtifactAttempts bounds how often UpsertArtifact retries a record that // concurrent commits keep moving. const upsertArtifactAttempts = 5 // UpsertArtifact records a. A storage path the record stops pointing at is // queued for deletion, since each fetch stores its own object and nothing else // would remove it. The write applies only if the record still holds the path // read before it, so commits racing on one artifact each queue the path they // replaced. // inUTC returns t in UTC. Artifacts store fetched_at and last_accessed_at in // UTC: lib/pq sends a time with its offset, a Postgres TIMESTAMP column keeps // only the wall clock, and lib/pq reads that back as UTC, so anything else // comes back shifted by the local offset. func inUTC(t sql.NullTime) sql.NullTime { if t.Valid { t.Time = t.Time.UTC() } return t } func (db *DB) UpsertArtifact(a *Artifact) error { for range upsertArtifactAttempts { var previous sql.NullString query := db.Rebind(`SELECT storage_path FROM artifacts WHERE version_purl = ? AND filename = ?`) if err := db.Get(&previous, query, a.VersionPURL, a.Filename); err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("reading artifact: %w", err) } applied, err := db.upsertArtifactFrom(a, previous) if err != nil { return err } if !applied { continue } if previous.Valid && previous.String != a.StoragePath.String { if err := db.QueuePendingDelete(previous.String); err != nil { return fmt.Errorf("queueing replaced artifact: %w", err) } } return nil } return errors.New("upserting artifact: record kept changing") } // upsertArtifactFrom writes a if the record is absent or still holds // previous, and reports whether it did. func (db *DB) upsertArtifactFrom(a *Artifact, previous sql.NullString) (bool, error) { now := time.Now() var query string if db.dialect == DialectPostgres { query = ` INSERT INTO artifacts (version_purl, filename, upstream_url, storage_path, content_hash, size, content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at, source, manifest) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) ON CONFLICT(version_purl, filename) DO UPDATE SET upstream_url = EXCLUDED.upstream_url, storage_path = EXCLUDED.storage_path, content_hash = EXCLUDED.content_hash, size = EXCLUDED.size, content_type = EXCLUDED.content_type, fetched_at = EXCLUDED.fetched_at, updated_at = EXCLUDED.updated_at, source = EXCLUDED.source, manifest = EXCLUDED.manifest WHERE artifacts.storage_path IS NOT DISTINCT FROM $15 ` } else { query = ` INSERT INTO artifacts (version_purl, filename, upstream_url, storage_path, content_hash, size, content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at, source, manifest) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(version_purl, filename) DO UPDATE SET upstream_url = excluded.upstream_url, storage_path = excluded.storage_path, content_hash = excluded.content_hash, size = excluded.size, content_type = excluded.content_type, fetched_at = excluded.fetched_at, updated_at = excluded.updated_at, source = excluded.source, manifest = excluded.manifest WHERE artifacts.storage_path IS ? ` } res, err := db.Exec(query, a.VersionPURL, a.Filename, a.UpstreamURL, a.StoragePath, a.ContentHash, a.Size, a.ContentType, inUTC(a.FetchedAt), a.HitCount, inUTC(a.LastAccessedAt), now, now, a.sourceOrUpstream(), a.Manifest, previous, ) if err != nil { return false, fmt.Errorf("upserting artifact: %w", err) } n, err := res.RowsAffected() if err != nil { return false, fmt.Errorf("upserting artifact: %w", err) } return n > 0, nil } func (db *DB) RecordArtifactHit(versionPURL, filename string) error { now := time.Now() if db.hits != nil { db.hits.add(hitKey{versionPURL, filename}, hitEntry{count: 1, last: now}) return nil } query := db.Rebind(` UPDATE artifacts SET hit_count = hit_count + 1, last_accessed_at = ?, updated_at = ? WHERE version_purl = ? AND filename = ? `) _, err := db.Exec(query, now.UTC(), now, versionPURL, filename) return err } func (db *DB) MarkArtifactCached(versionPURL, filename, storagePath, contentHash string, size int64, contentType string) error { now := time.Now() query := db.Rebind(` UPDATE artifacts SET storage_path = ?, content_hash = ?, size = ?, content_type = ?, fetched_at = ?, updated_at = ? WHERE version_purl = ? AND filename = ? `) _, err := db.Exec(query, storagePath, contentHash, size, contentType, now.UTC(), now, versionPURL, filename) return err } // Cache management queries func (db *DB) GetLeastRecentlyUsedArtifacts(limit int) ([]Artifact, error) { var artifacts []Artifact query := db.Rebind(` SELECT id, version_purl, filename, upstream_url, storage_path, content_hash, size, content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at, source FROM artifacts WHERE storage_path IS NOT NULL AND source <> 'local' ORDER BY last_accessed_at ASC NULLS FIRST LIMIT ? `) err := db.Select(&artifacts, query, limit) if err != nil { return nil, err } return artifacts, nil } func (db *DB) GetTotalCacheSize() (int64, error) { var total sql.NullInt64 err := db.Get(&total, `SELECT SUM(size) FROM artifacts WHERE storage_path IS NOT NULL AND source <> 'local'`) if err != nil { return 0, err } if !total.Valid { return 0, nil } return total.Int64, nil } func (db *DB) GetCachedArtifactCount() (int64, error) { var count int64 err := db.Get(&count, `SELECT COUNT(*) FROM artifacts WHERE storage_path IS NOT NULL`) return count, err } // ClearArtifactCache marks an artifact uncached if its record still points at // storagePath, and reports whether it did. A record a newer fetch has moved // elsewhere is left alone, so a clear never orphans the object that fetch // committed. func (db *DB) ClearArtifactCache(versionPURL, filename, storagePath string) (bool, error) { query := db.Rebind(` UPDATE artifacts SET storage_path = NULL, content_hash = NULL, size = NULL, content_type = NULL, fetched_at = NULL, updated_at = ? WHERE version_purl = ? AND filename = ? AND storage_path = ? `) res, err := db.Exec(query, time.Now(), versionPURL, filename, storagePath) if err != nil { return false, err } n, err := res.RowsAffected() return n > 0, err } // DiscardArtifact clears the record as ClearArtifactCache does and queues // storagePath for deletion, for callers that must not delete an object another // request may be reading. func (db *DB) DiscardArtifact(versionPURL, filename, storagePath string) error { cleared, err := db.ClearArtifactCache(versionPURL, filename, storagePath) if err != nil || !cleared { return err } return db.QueuePendingDelete(storagePath) } // QueuePendingDelete queues a storage path no record points at any more. // Queueing a path again, as reclaim does after a failed delete, restarts its // grace period and moves it behind the rest. func (db *DB) QueuePendingDelete(path string) error { _, err := db.Exec(db.Rebind(queuePendingDeleteQuery), path, time.Now().UTC()) return err } // queuePendingDeleteQuery queues a path, or moves an already queued one to // the back of the queue. It takes the path and the time it is queued at. const queuePendingDeleteQuery = ` INSERT INTO pending_deletes (path, queued_at) VALUES (?, ?) ON CONFLICT(path) DO UPDATE SET queued_at = excluded.queued_at` // GetDuePendingDeletes returns up to limit paths queued before cutoff, oldest // first. A path a record points at again is skipped. func (db *DB) GetDuePendingDeletes(cutoff time.Time, limit int) ([]string, error) { var paths []string query := db.Rebind(` SELECT path FROM pending_deletes WHERE queued_at < ? AND NOT EXISTS (SELECT 1 FROM artifacts WHERE artifacts.storage_path = pending_deletes.path) ORDER BY queued_at LIMIT ? `) if err := db.Select(&paths, query, cutoff.UTC(), limit); err != nil { return nil, err } return paths, nil } // RemovePendingDelete drops a deleted path from the queue. func (db *DB) RemovePendingDelete(path string) error { _, err := db.Exec(db.Rebind(`DELETE FROM pending_deletes WHERE path = ?`), path) return err } // Stats queries type CacheStats struct { TotalPackages int64 TotalVersions int64 TotalArtifacts int64 TotalSize int64 TotalHits int64 EcosystemCounts map[string]int64 } func (db *DB) GetCacheStats() (*CacheStats, error) { stats := &CacheStats{ EcosystemCounts: make(map[string]int64), } if err := db.Get(&stats.TotalPackages, `SELECT COUNT(*) FROM packages`); err != nil { return nil, err } if err := db.Get(&stats.TotalVersions, `SELECT COUNT(*) FROM versions`); err != nil { return nil, err } // Check if artifacts table exists (might not if using git-pkgs db without proxy tables) hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } if hasArtifacts { row := db.QueryRow(` SELECT COUNT(*), COALESCE(SUM(size), 0) FROM artifacts WHERE storage_path IS NOT NULL `) if err := row.Scan(&stats.TotalArtifacts, &stats.TotalSize); err != nil { return nil, err } var totalHits sql.NullInt64 if err := db.Get(&totalHits, `SELECT SUM(hit_count) FROM artifacts`); err != nil { return nil, err } if totalHits.Valid { stats.TotalHits = totalHits.Int64 } } rows, err := db.Query(`SELECT ecosystem, COUNT(*) FROM packages GROUP BY ecosystem`) if err != nil { return nil, err } defer func() { _ = rows.Close() }() for rows.Next() { var ecosystem string var count int64 if err := rows.Scan(&ecosystem, &count); err != nil { return nil, err } stats.EcosystemCounts[ecosystem] = count } return stats, rows.Err() } type PopularPackage struct { Ecosystem string `db:"ecosystem"` Name string `db:"name"` Hits int64 `db:"hits"` Size int64 `db:"size"` } func (db *DB) GetMostPopularPackages(limit int) ([]PopularPackage, error) { // Check if artifacts table exists hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } if !hasArtifacts { return nil, nil } var packages []PopularPackage query := db.Rebind(` SELECT p.ecosystem, p.name, COALESCE(SUM(a.hit_count), 0) as hits, COALESCE(SUM(a.size), 0) as size FROM packages p JOIN versions v ON v.package_purl = p.purl JOIN artifacts a ON a.version_purl = v.purl WHERE a.storage_path IS NOT NULL GROUP BY p.purl, p.ecosystem, p.name ORDER BY hits DESC LIMIT ? `) err = db.Select(&packages, query, limit) if err != nil { return nil, err } return packages, nil } type RecentPackage struct { Ecosystem string `db:"ecosystem"` Name string `db:"name"` VersionPURL string `db:"version_purl"` CachedAt time.Time `db:"fetched_at"` Size int64 `db:"size"` // Version is derived from VersionPURL rather than selected, so that the // PURL percent-encoding is decoded (e.g. "%2B" back to "+"). Version string `db:"-"` } func (db *DB) GetRecentlyCachedPackages(limit int) ([]RecentPackage, error) { // Check if artifacts table exists hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } if !hasArtifacts { return nil, nil } var packages []RecentPackage // There is no separate version column, so the full version PURL is selected // and the version is decoded from it in Go. query := db.Rebind(` SELECT p.ecosystem, p.name, v.purl as version_purl, a.fetched_at, COALESCE(a.size, 0) as size FROM artifacts a JOIN versions v ON v.purl = a.version_purl JOIN packages p ON p.purl = v.package_purl WHERE a.storage_path IS NOT NULL AND a.fetched_at IS NOT NULL ORDER BY a.fetched_at DESC LIMIT ? `) err = db.Select(&packages, query, limit) if err != nil { return nil, err } for i := range packages { packages[i].Version = VersionFromPURL(packages[i].VersionPURL) } return packages, nil } type SearchResult struct { Ecosystem string `db:"ecosystem"` Name string `db:"name"` LatestVersion sql.NullString `db:"latest_version"` License sql.NullString `db:"license"` Hits int64 `db:"hits"` Size int64 `db:"size"` CachedAt sql.NullString `db:"cached_at"` } func (db *DB) SearchPackages(query string, ecosystem string, limit int, offset int) ([]SearchResult, error) { // Check if artifacts table exists hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } if !hasArtifacts { return nil, nil } var results []SearchResult searchPattern := "%" + query + "%" var sqlQuery string var args []any if ecosystem != "" { sqlQuery = db.Rebind(` SELECT p.ecosystem, p.name, p.latest_version, p.license, COALESCE(SUM(a.hit_count), 0) as hits, COALESCE(SUM(a.size), 0) as size, MAX(a.fetched_at) as cached_at FROM packages p LEFT JOIN versions v ON v.package_purl = p.purl LEFT JOIN artifacts a ON a.version_purl = v.purl WHERE p.name LIKE ? AND p.ecosystem = ? AND a.storage_path IS NOT NULL GROUP BY p.purl, p.ecosystem, p.name, p.latest_version, p.license ORDER BY hits DESC LIMIT ? OFFSET ? `) args = []any{searchPattern, ecosystem, limit, offset} } else { sqlQuery = db.Rebind(` SELECT p.ecosystem, p.name, p.latest_version, p.license, COALESCE(SUM(a.hit_count), 0) as hits, COALESCE(SUM(a.size), 0) as size, MAX(a.fetched_at) as cached_at FROM packages p LEFT JOIN versions v ON v.package_purl = p.purl LEFT JOIN artifacts a ON a.version_purl = v.purl WHERE p.name LIKE ? AND a.storage_path IS NOT NULL GROUP BY p.purl, p.ecosystem, p.name, p.latest_version, p.license ORDER BY hits DESC LIMIT ? OFFSET ? `) args = []any{searchPattern, limit, offset} } err = db.Select(&results, sqlQuery, args...) if err != nil { return nil, err } return results, nil } func (db *DB) CountSearchResults(query string, ecosystem string) (int64, error) { hasArtifacts, err := db.HasTable("artifacts") if err != nil { return 0, err } if !hasArtifacts { return 0, nil } searchPattern := "%" + query + "%" var sqlQuery string var args []any if ecosystem != "" { sqlQuery = db.Rebind(` SELECT COUNT(DISTINCT p.purl) FROM packages p LEFT JOIN versions v ON v.package_purl = p.purl LEFT JOIN artifacts a ON a.version_purl = v.purl WHERE p.name LIKE ? AND p.ecosystem = ? AND a.storage_path IS NOT NULL `) args = []any{searchPattern, ecosystem} } else { sqlQuery = db.Rebind(` SELECT COUNT(DISTINCT p.purl) FROM packages p LEFT JOIN versions v ON v.package_purl = p.purl LEFT JOIN artifacts a ON a.version_purl = v.purl WHERE p.name LIKE ? AND a.storage_path IS NOT NULL `) args = []any{searchPattern} } var count int64 err = db.Get(&count, sqlQuery, args...) return count, err } // Vulnerability queries func (db *DB) GetVulnerabilitiesForPackage(ecosystem, name string) ([]Vulnerability, error) { var vulns []Vulnerability query := db.Rebind(` SELECT id, vuln_id, ecosystem, package_name, severity, summary, fixed_version, cvss_score, "references", fetched_at, created_at, updated_at FROM vulnerabilities WHERE ecosystem = ? AND package_name = ? ORDER BY cvss_score DESC NULLS LAST `) err := db.Select(&vulns, query, ecosystem, name) if err != nil { return nil, err } return vulns, nil } func (db *DB) UpsertVulnerability(v *Vulnerability) error { now := time.Now() var query string if db.dialect == DialectPostgres { query = ` INSERT INTO vulnerabilities (vuln_id, ecosystem, package_name, severity, summary, fixed_version, cvss_score, "references", fetched_at, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT(vuln_id, ecosystem, package_name) DO UPDATE SET severity = EXCLUDED.severity, summary = EXCLUDED.summary, fixed_version = EXCLUDED.fixed_version, cvss_score = EXCLUDED.cvss_score, "references" = EXCLUDED."references", fetched_at = EXCLUDED.fetched_at, updated_at = EXCLUDED.updated_at ` } else { query = ` INSERT INTO vulnerabilities (vuln_id, ecosystem, package_name, severity, summary, fixed_version, cvss_score, "references", fetched_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(vuln_id, ecosystem, package_name) DO UPDATE SET severity = excluded.severity, summary = excluded.summary, fixed_version = excluded.fixed_version, cvss_score = excluded.cvss_score, "references" = excluded."references", fetched_at = excluded.fetched_at, updated_at = excluded.updated_at ` } _, err := db.Exec(query, v.VulnID, v.Ecosystem, v.PackageName, v.Severity, v.Summary, v.FixedVersion, v.CVSSScore, v.References, v.FetchedAt, now, now, ) if err != nil { return fmt.Errorf("upserting vulnerability: %w", err) } return nil } func (db *DB) DeleteVulnerabilitiesForPackage(ecosystem, name string) error { query := db.Rebind(`DELETE FROM vulnerabilities WHERE ecosystem = ? AND package_name = ?`) _, err := db.Exec(query, ecosystem, name) return err } func (db *DB) GetVulnsSyncedAt(ecosystem, name string) (time.Time, error) { var syncedAt sql.NullTime query := db.Rebind(`SELECT vulns_synced_at FROM packages WHERE ecosystem = ? AND name = ?`) err := db.Get(&syncedAt, query, ecosystem, name) if err == sql.ErrNoRows || !syncedAt.Valid { return time.Time{}, nil } if err != nil { return time.Time{}, err } return syncedAt.Time, nil } func (db *DB) SetVulnsSyncedAt(ecosystem, name string) error { now := time.Now() query := db.Rebind(`UPDATE packages SET vulns_synced_at = ?, updated_at = ? WHERE ecosystem = ? AND name = ?`) _, err := db.Exec(query, now, now, ecosystem, name) return err } func (db *DB) GetPackagesNeedingVulnSync(limit int, minAge time.Duration) ([]Package, error) { cutoff := time.Now().Add(-minAge) var packages []Package query := db.Rebind(` SELECT id, purl, ecosystem, name, latest_version, license, description, homepage, repository_url, registry_url, supplier_name, supplier_type, source, enriched_at, vulns_synced_at, created_at, updated_at FROM packages WHERE vulns_synced_at IS NULL OR vulns_synced_at < ? ORDER BY vulns_synced_at ASC NULLS FIRST LIMIT ? `) err := db.Select(&packages, query, cutoff, limit) if err != nil { return nil, err } return packages, nil } func (db *DB) GetVulnCountForPackage(ecosystem, name string) (int64, error) { var count int64 query := db.Rebind(`SELECT COUNT(*) FROM vulnerabilities WHERE ecosystem = ? AND package_name = ?`) err := db.Get(&count, query, ecosystem, name) return count, err } // Enrichment stats type EnrichmentStats struct { TotalPackages int64 EnrichedPackages int64 VulnSyncedPackages int64 TotalVulnerabilities int64 CriticalVulns int64 HighVulns int64 MediumVulns int64 LowVulns int64 } func (db *DB) GetEnrichmentStats() (*EnrichmentStats, error) { stats := &EnrichmentStats{} if err := db.Get(&stats.TotalPackages, `SELECT COUNT(*) FROM packages`); err != nil { return nil, err } if err := db.Get(&stats.EnrichedPackages, `SELECT COUNT(*) FROM packages WHERE enriched_at IS NOT NULL`); err != nil { return nil, err } if err := db.Get(&stats.VulnSyncedPackages, `SELECT COUNT(*) FROM packages WHERE vulns_synced_at IS NOT NULL`); err != nil { return nil, err } // Check if vulnerabilities table exists hasVulns, err := db.HasTable("vulnerabilities") if err != nil { return nil, err } if hasVulns { if err := db.Get(&stats.TotalVulnerabilities, `SELECT COUNT(*) FROM vulnerabilities`); err != nil { return nil, err } if err := db.Get(&stats.CriticalVulns, `SELECT COUNT(*) FROM vulnerabilities WHERE severity = 'critical'`); err != nil { return nil, err } if err := db.Get(&stats.HighVulns, `SELECT COUNT(*) FROM vulnerabilities WHERE severity = 'high'`); err != nil { return nil, err } if err := db.Get(&stats.MediumVulns, `SELECT COUNT(*) FROM vulnerabilities WHERE severity = 'medium'`); err != nil { return nil, err } if err := db.Get(&stats.LowVulns, `SELECT COUNT(*) FROM vulnerabilities WHERE severity = 'low'`); err != nil { return nil, err } } return stats, nil } type PackageListItem struct { Ecosystem string `db:"ecosystem"` Name string `db:"name"` LatestVersion sql.NullString `db:"latest_version"` License sql.NullString `db:"license"` Hits int64 `db:"hits"` Size int64 `db:"size"` CachedAt sql.NullString `db:"cached_at"` VulnCount int64 `db:"vuln_count"` } func (db *DB) ListCachedPackages(ecosystem string, sortBy string, limit int, offset int) ([]PackageListItem, error) { hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } if !hasArtifacts { return nil, nil } hasVulns, err := db.HasTable("vulnerabilities") if err != nil { return nil, err } vulnJoin := "" vulnSelect := "0 as vuln_count" if hasVulns { vulnJoin = "LEFT JOIN (SELECT ecosystem, package_name, COUNT(*) as vuln_count FROM vulnerabilities GROUP BY ecosystem, package_name) v ON v.ecosystem = p.ecosystem AND v.package_name = p.name" vulnSelect = "COALESCE(v.vuln_count, 0) as vuln_count" } orderClause := "ORDER BY hits DESC" switch sortBy { case "name": orderClause = "ORDER BY p.name ASC" case "size": orderClause = "ORDER BY size DESC" case "cached_at": orderClause = "ORDER BY cached_at DESC" case "ecosystem": orderClause = "ORDER BY p.ecosystem ASC, p.name ASC" case "vulns": orderClause = "ORDER BY vuln_count DESC, p.name ASC" } whereClause := "WHERE a.storage_path IS NOT NULL" args := []any{} if ecosystem != "" { whereClause += " AND p.ecosystem = ?" args = append(args, ecosystem) } groupByClause := "GROUP BY p.purl, p.ecosystem, p.name, p.latest_version, p.license" if hasVulns { groupByClause += ", v.vuln_count" } query := fmt.Sprintf(` SELECT p.ecosystem, p.name, p.latest_version, p.license, COALESCE(SUM(a.hit_count), 0) as hits, COALESCE(SUM(a.size), 0) as size, MAX(a.fetched_at) as cached_at, %s FROM packages p JOIN versions v2 ON v2.package_purl = p.purl JOIN artifacts a ON a.version_purl = v2.purl %s %s %s %s LIMIT ? OFFSET ? `, vulnSelect, vulnJoin, whereClause, groupByClause, orderClause) args = append(args, limit, offset) query = db.Rebind(query) var packages []PackageListItem err = db.Select(&packages, query, args...) if err != nil { return nil, err } return packages, nil } // CountCachedPackagesByEcosystem counts packages with at least one stored artifact. func (db *DB) CountCachedPackagesByEcosystem() (map[string]int64, error) { hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } counts := make(map[string]int64) if !hasArtifacts { return counts, nil } rows, err := db.Query(` SELECT p.ecosystem, COUNT(DISTINCT p.purl) FROM packages p JOIN versions v ON v.package_purl = p.purl JOIN artifacts a ON a.version_purl = v.purl WHERE a.storage_path IS NOT NULL GROUP BY p.ecosystem `) if err != nil { return nil, err } defer func() { _ = rows.Close() }() for rows.Next() { var ecosystem string var count int64 if err := rows.Scan(&ecosystem, &count); err != nil { return nil, err } counts[ecosystem] = count } return counts, rows.Err() } func (db *DB) CountCachedPackages(ecosystem string) (int64, error) { hasArtifacts, err := db.HasTable("artifacts") if err != nil { return 0, err } if !hasArtifacts { return 0, nil } whereClause := "WHERE a.storage_path IS NOT NULL" args := []any{} if ecosystem != "" { whereClause += " AND p.ecosystem = ?" args = append(args, ecosystem) } query := fmt.Sprintf(` SELECT COUNT(DISTINCT p.purl) FROM packages p JOIN versions v ON v.package_purl = p.purl JOIN artifacts a ON a.version_purl = v.purl %s `, whereClause) query = db.Rebind(query) var count int64 err = db.Get(&count, query, args...) return count, err } // Metadata cache queries func (db *DB) GetMetadataCache(ecosystem, name string) (*MetadataCacheEntry, error) { var entry MetadataCacheEntry query := db.Rebind(` SELECT id, ecosystem, name, storage_path, etag, link, content_type, content_encoding, content_digest, size, last_modified, fetched_at, created_at, updated_at FROM metadata_cache WHERE ecosystem = ? AND name = ? `) err := db.Get(&entry, query, ecosystem, name) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &entry, nil } func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { now := time.Now() // fetched_at is a TIMESTAMP without zone on Postgres, and lib/pq reads it // back labelled UTC, so it has to be stored in UTC. last_modified already // is, since it comes from an HTTP date. fetchedAt := entry.FetchedAt if fetchedAt.Valid { fetchedAt.Time = fetchedAt.Time.UTC() } var query string if db.dialect == DialectPostgres { query = ` INSERT INTO metadata_cache (ecosystem, name, storage_path, etag, link, content_type, content_encoding, content_digest, size, last_modified, fetched_at, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT(ecosystem, name) DO UPDATE SET storage_path = EXCLUDED.storage_path, etag = EXCLUDED.etag, link = EXCLUDED.link, content_type = EXCLUDED.content_type, content_encoding = EXCLUDED.content_encoding, content_digest = EXCLUDED.content_digest, size = EXCLUDED.size, last_modified = EXCLUDED.last_modified, fetched_at = EXCLUDED.fetched_at, updated_at = EXCLUDED.updated_at ` } else { query = ` INSERT INTO metadata_cache (ecosystem, name, storage_path, etag, link, content_type, content_encoding, content_digest, size, last_modified, fetched_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(ecosystem, name) DO UPDATE SET storage_path = excluded.storage_path, etag = excluded.etag, link = excluded.link, content_type = excluded.content_type, content_encoding = excluded.content_encoding, content_digest = excluded.content_digest, size = excluded.size, last_modified = excluded.last_modified, fetched_at = excluded.fetched_at, updated_at = excluded.updated_at ` } _, err := db.Exec(query, entry.Ecosystem, entry.Name, entry.StoragePath, entry.ETag, entry.Link, entry.ContentType, entry.ContentEncoding, entry.ContentDigest, entry.Size, entry.LastModified, fetchedAt, now, now, ) if err != nil { return fmt.Errorf("upserting metadata cache: %w", err) } return nil } // Analytics queries // EcosystemStats aggregates cache and download activity for one ecosystem. // // DownloadedBytes is the accumulated download volume: every cache hit on an // artifact served its full size, so the sum of hit_count * size is the number // of bytes the proxy has handed to clients from cache for this ecosystem. type EcosystemStats struct { Ecosystem string `db:"ecosystem"` Packages int64 `db:"packages"` Versions int64 `db:"versions"` Artifacts int64 `db:"artifacts"` CacheSize int64 `db:"cache_size"` Downloads int64 `db:"downloads"` DownloadedBytes int64 `db:"downloaded_bytes"` } // GetEcosystemStats returns per-ecosystem cache and download totals, ordered by // accumulated download volume descending. Ecosystems with rows in packages but // nothing cached are included with zeroed artifact counters. // // Artifacts evicted from the cache no longer contribute: eviction clears the // size column, so their historical hits drop out of the accumulated total. func (db *DB) GetEcosystemStats() ([]EcosystemStats, error) { byEcosystem := make(map[string]*EcosystemStats) get := func(ecosystem string) *EcosystemStats { if s, ok := byEcosystem[ecosystem]; ok { return s } s := &EcosystemStats{Ecosystem: ecosystem} byEcosystem[ecosystem] = s return s } if err := db.eachCount(`SELECT ecosystem, COUNT(*) FROM packages GROUP BY ecosystem`, func(ecosystem string, n int64) { get(ecosystem).Packages = n }); err != nil { return nil, err } // Left joined and bucketed for the same reason as artifacts below: a version // whose package row is missing would otherwise vanish here while still // counting in GetCacheStats' COUNT(*), leaving the two unable to reconcile. if err := db.eachCount(` SELECT COALESCE(p.ecosystem, '`+unattributedEcosystem+`'), COUNT(*) FROM versions v LEFT JOIN packages p ON p.purl = v.package_purl GROUP BY COALESCE(p.ecosystem, '`+unattributedEcosystem+`') `, func(ecosystem string, n int64) { get(ecosystem).Versions = n }); err != nil { return nil, err } // The artifacts table is proxy-specific: a database inherited from // git-pkgs carries packages and versions without it. hasArtifacts, err := db.HasTable("artifacts") if err != nil { return nil, err } if hasArtifacts { if err := db.eachArtifactStat(get); err != nil { return nil, err } } stats := make([]EcosystemStats, 0, len(byEcosystem)) for _, s := range byEcosystem { stats = append(stats, *s) } stats = mergeAliasedEcosystems(stats) sortEcosystemStats(stats) return stats, nil } // mergeAliasedEcosystems combines rows whose ecosystem names normalize to the // same canonical name. The proxy writes "gem" and git-pkgs writes "rubygems", // so a database that has seen both carries two rows for one ecosystem; left // split they would render as two table rows and two chart slices, each with // half the real share. // // The surviving row keeps the raw spelling of whichever input held the most // cached bytes, because that string is what the UI filters and links by: the // packages table stores the raw value, so substituting the canonical name would // produce links that match nothing. That comparison is against each input's own // size, not the running total, which would otherwise let the first spelling win // simply by being merged into first. func mergeAliasedEcosystems(stats []EcosystemStats) []EcosystemStats { merged := make(map[string]*EcosystemStats, len(stats)) largest := make(map[string]int64, len(stats)) order := make([]string, 0, len(stats)) for i := range stats { key := purl.NormalizeEcosystem(stats[i].Ecosystem) into, ok := merged[key] if !ok { row := stats[i] merged[key] = &row largest[key] = stats[i].CacheSize order = append(order, key) continue } // The name tiebreak matters: GetEcosystemStats builds its input by // ranging a map, so without it two spellings of equal size would swap // between refreshes and flip the row's badge and filter link. if stats[i].CacheSize > largest[key] || (stats[i].CacheSize == largest[key] && stats[i].Ecosystem < into.Ecosystem) { largest[key] = stats[i].CacheSize into.Ecosystem = stats[i].Ecosystem } into.Packages += stats[i].Packages into.Versions += stats[i].Versions into.Artifacts += stats[i].Artifacts into.CacheSize += stats[i].CacheSize into.Downloads += stats[i].Downloads into.DownloadedBytes += stats[i].DownloadedBytes } out := make([]EcosystemStats, 0, len(order)) for _, key := range order { out = append(out, *merged[key]) } return out } // unattributedEcosystem collects cached artifacts whose version or package row // is missing. Nothing enforces that link at the schema level, so an inner join // would silently drop such rows and leave the per-ecosystem totals short of // GetTotalCacheSize — two numbers that sit side by side in the UI and in // Grafana. Bucketing them keeps the two reconcilable. const unattributedEcosystem = "unattributed" func (db *DB) eachArtifactStat(get func(string) *EcosystemStats) error { rows, err := db.Query(` SELECT COALESCE(p.ecosystem, '` + unattributedEcosystem + `'), COUNT(*), COALESCE(SUM(a.size), 0), COALESCE(SUM(a.hit_count), 0), COALESCE(SUM(a.hit_count * a.size), 0) FROM artifacts a LEFT JOIN versions v ON v.purl = a.version_purl LEFT JOIN packages p ON p.purl = v.package_purl WHERE a.storage_path IS NOT NULL GROUP BY COALESCE(p.ecosystem, '` + unattributedEcosystem + `') `) if err != nil { return err } defer func() { _ = rows.Close() }() for rows.Next() { var ecosystem string var artifacts, cacheSize, downloads, downloadedBytes int64 if err := rows.Scan(&ecosystem, &artifacts, &cacheSize, &downloads, &downloadedBytes); err != nil { return err } s := get(ecosystem) s.Artifacts = artifacts s.CacheSize = cacheSize s.Downloads = downloads s.DownloadedBytes = downloadedBytes } return rows.Err() } // eachCount runs a two-column "group by" query and hands each (key, count) pair to fn. func (db *DB) eachCount(query string, fn func(key string, n int64)) error { rows, err := db.Query(query) if err != nil { return err } defer func() { _ = rows.Close() }() for rows.Next() { var key string var n int64 if err := rows.Scan(&key, &n); err != nil { return err } fn(key, n) } return rows.Err() } // sortEcosystemStats orders by accumulated download volume, then by cache size, // then by name, so that ecosystems with no traffic yet still sort predictably. func sortEcosystemStats(stats []EcosystemStats) { sort.Slice(stats, func(i, j int) bool { a, b := stats[i], stats[j] switch { case a.DownloadedBytes != b.DownloadedBytes: return a.DownloadedBytes > b.DownloadedBytes case a.CacheSize != b.CacheSize: return a.CacheSize > b.CacheSize default: return a.Ecosystem < b.Ecosystem } }) }