package database import ( "database/sql" "fmt" "time" ) // LocalArtifact is a pushed artifact with the integrity of its version. type LocalArtifact struct { Artifact Integrity sql.NullString `db:"integrity"` } // GetLocalArtifacts returns the artifacts pushed to the proxy for a package, // ordered by version and filename. func (db *DB) GetLocalArtifacts(packagePURL string) ([]LocalArtifact, error) { var artifacts []LocalArtifact query := db.Rebind(` SELECT a.id, a.version_purl, a.filename, a.upstream_url, a.storage_path, a.content_hash, a.size, a.content_type, a.fetched_at, a.hit_count, a.last_accessed_at, a.created_at, a.updated_at, a.source, a.manifest, v.integrity FROM artifacts a JOIN versions v ON v.purl = a.version_purl WHERE v.package_purl = ? AND a.source = 'local' ORDER BY a.version_purl, a.filename `) if err := db.Select(&artifacts, query, packagePURL); err != nil { return nil, err } return artifacts, nil } // DeleteLocalArtifacts removes the pushed artifacts of a version, or only the // one named filename when it is set, and queues their stored objects for // deletion in the same transaction, so a failed queue keeps the records // instead of orphaning the objects. Artifacts cached from an upstream are left // alone. It returns how many records it removed. func (db *DB) DeleteLocalArtifacts(versionPURL, filename string) (int, error) { tx, err := db.Beginx() if err != nil { return 0, err } defer func() { _ = tx.Rollback() }() where := "version_purl = ? AND source = 'local'" args := []any{versionPURL} if filename != "" { where += " AND filename = ?" args = append(args, filename) } var paths []string selectQuery := db.Rebind(`SELECT storage_path FROM artifacts WHERE ` + where + ` AND storage_path IS NOT NULL`) if err := tx.Select(&paths, selectQuery, args...); err != nil { return 0, fmt.Errorf("reading local artifacts: %w", err) } res, err := tx.Exec(db.Rebind(`DELETE FROM artifacts WHERE `+where), args...) if err != nil { return 0, fmt.Errorf("deleting local artifacts: %w", err) } deleted, err := res.RowsAffected() if err != nil { return 0, err } queuedAt := time.Now().UTC() for _, path := range paths { if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), path, queuedAt); err != nil { return 0, fmt.Errorf("queueing %s for deletion: %w", path, err) } } return int(deleted), tx.Commit() } // RecordLocalArtifact records a pushed artifact in one transaction: the // package if it is missing, the version, and the artifact. An artifact cached // from an upstream under the same name is replaced and its stored object // queued for deletion; a pushed one is never replaced, and the whole record // is refused with ErrLocalArtifactExists, so of two concurrent pushes of the // same file exactly one wins. // // When oneFilePerVersion is false, an existing version record is left as it // is. Otherwise its integrity is set to v.Integrity, and the push is refused // if the version already holds a pushed artifact under any filename, since // aliases of one name (Composer's are case-insensitive) share the version // record and its integrity. The version upsert locks that record first, so // concurrent pushes of aliases see each other. func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, oneFilePerVersion bool, a *Artifact) error { local := *a local.Source = SourceLocal return db.recordArtifact(&local, func(tx execer) error { now := time.Now() if _, err := tx.Exec(db.Rebind(` INSERT INTO packages (purl, ecosystem, name, created_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(purl) DO NOTHING `), pkg.PURL, pkg.Ecosystem, pkg.Name, now, now); err != nil { return fmt.Errorf("inserting package: %w", err) } versionConflict := `DO NOTHING` if oneFilePerVersion { versionConflict = `DO UPDATE SET integrity = excluded.integrity, updated_at = excluded.updated_at` } if _, err := tx.Exec(db.Rebind(` INSERT INTO versions (purl, package_purl, integrity, created_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(purl) `+versionConflict, ), v.PURL, v.PackagePURL, v.Integrity, now, now); err != nil { return fmt.Errorf("recording version: %w", err) } if oneFilePerVersion { var pushed int if err := tx.Get(&pushed, db.Rebind(` SELECT COUNT(*) FROM artifacts WHERE version_purl = ? AND source = ? `), v.PURL, SourceLocal); err != nil { return fmt.Errorf("checking pushed artifacts: %w", err) } if pushed > 0 { return ErrLocalArtifactExists } } return nil }, ErrLocalArtifactExists) }