Skip to content
go-libfossil Artifact [4b0c015645]
Login

Artifact [4b0c015645]

Check-ins Using Download Hex

Artifact 4b0c0156457f70c0e3409321f58cc6317e26dae84d9776057c2df38b1bcb6c23:

package blob

import (
	"fmt"

	"github.com/danmestas/go-libfossil/db"
	"github.com/danmestas/go-libfossil/internal/delta"
	libfossil "github.com/danmestas/go-libfossil/internal/fsltype"
	"github.com/danmestas/go-libfossil/internal/hash"
)

// existingName reports the name content is already stored under, if any.
// uuid is content's name under the repo's current policy; when that name is
// absent and the policy still honours legacy names, the other algorithm's
// name counts too — an artifact stored before the repo moved to SHA3 keeps
// the name every existing manifest already refers to, instead of being
// stored a second time and churning every F-card that mentions it.
//
// Only a stored blob qualifies. A phantom is a name the repo knows of but
// has no content for, so reusing one would report content as stored that
// isn't; callers fall through to writeRow, which fills it.
func existingName(q db.Querier, naming hash.Naming, uuid string, content []byte) (libfossil.FslID, string, bool) {
	if rid, ok := Exists(q, uuid); ok && !isPhantom(q, rid) {
		return rid, uuid, true
	}
	// A phantom under the primary name does not end the search: fossil's
	// content_put_ex consults the other algorithm's name first and uses a
	// stored blob there, so content already held under it is not stored twice
	// just because the primary name is waiting for it.
	if !naming.ReuseLegacy {
		return 0, "", false
	}
	legacy := naming.New.Other().Hash(content)
	rid, ok := Exists(q, legacy)
	if !ok {
		return 0, "", false
	}
	if isPhantom(q, rid) {
		return 0, "", false
	}
	return rid, legacy, true
}

// isPhantom reports whether rid is a placeholder with no content. A row
// that cannot be read counts as one, so callers never reuse it.
func isPhantom(q db.Querier, rid libfossil.FslID) bool {
	var size int64
	err := q.QueryRow("SELECT size FROM blob WHERE rid=?", rid).Scan(&size)
	return err != nil || size == -1
}

// writeRow stores a blob row for uuid and returns its rid. If uuid is
// already a phantom the row is filled in place, so every reference to its
// rid starts resolving; otherwise a new row is inserted. Mirrors the
// phantom branch of Fossil's content_put_ex (src/content.c:557-620).
func writeRow(q db.Querier, uuid string, size int64, compressed []byte) (libfossil.FslID, error) {
	if q == nil {
		panic("blob.writeRow: q must not be nil")
	}
	if size < 0 {
		panic("blob.writeRow: size must not be negative")
	}
	if len(compressed) == 0 {
		panic("blob.writeRow: compressed must not be empty")
	}
	// Promote before the name is verified against content, as fossil's
	// content_put_ex does: a received name is trusted until Expand checks it.
	if err := hash.PromoteFor(q, uuid); err != nil {
		return 0, fmt.Errorf("hash policy: %w", err)
	}
	// The receipt is made only once there is content to store under it, as
	// content_put_ex calls content_rcvid_init after its already-held check.
	rcvid, err := db.ReceiptID(q)
	if err != nil {
		return 0, fmt.Errorf("receipt: %w", err)
	}
	if rid, ok := Exists(q, uuid); ok {
		if _, err := q.Exec("UPDATE blob SET rcvid=?, size=? WHERE rid=?",
			rcvid, size, rid); err != nil {
			return 0, fmt.Errorf("fill phantom: %w", err)
		}
		if err := WriteStored(q, rid, compressed); err != nil {
			return 0, fmt.Errorf("fill phantom: %w", err)
		}
		if _, err := q.Exec("DELETE FROM phantom WHERE rid=?", rid); err != nil {
			return 0, fmt.Errorf("clear phantom: %w", err)
		}
		return rid, nil
	}
	return InsertStored(q, uuid, size, compressed, rcvid)
}

func Store(q db.Querier, content []byte) (rid libfossil.FslID, uuid string, err error) {
	if q == nil {
		panic("blob.Store: q must not be nil")
	}
	// Zero-length content is legal: the empty artifact is a normal Fossil
	// blob with a well-known hash (SHA1 da39a3ee...). A caller may pass
	// nil for it (an unread/empty file) rather than remembering to
	// substitute []byte{}; treat the two identically.
	if content == nil {
		content = []byte{}
	}
	defer func() {
		if err == nil && rid <= 0 {
			panic("blob.Store: postcondition violated: rid <= 0 with no error")
		}
	}()

	naming, err := hash.NamingFor(q)
	if err != nil {
		return 0, "", fmt.Errorf("blob: %w", err)
	}
	uuid = naming.New.Hash(content)

	if rid, name, ok := existingName(q, naming, uuid, content); ok {
		return rid, name, nil
	}
	rid, err = writeVerified(q, naming.New, uuid, content)
	if err != nil {
		return 0, "", fmt.Errorf("blob.Store: %w", err)
	}
	return rid, uuid, nil
}

// Outcome is what StoreUnderPolicy did with an artifact.
type Outcome int

const (
	Found         Outcome = iota // already stored under its name
	Inserted                     // stored as a new row
	FilledPhantom                // stored into the phantom awaiting it
)

// StoreUnderPolicy stores content under the name the repository's hash
// policy gives it, never reusing a copy held under the other algorithm's
// name: an S card names its sub-manifest "under the repository hash policy"
// (draft-fossil-submanifests-00 §6.2), so two writers name a tree alike
// whatever legacy copies either holds. It reports whether the row was
// found, inserted, or filled a phantom.
func StoreUnderPolicy(q db.Querier, content []byte) (rid libfossil.FslID, uuid string, outcome Outcome, err error) {
	if q == nil {
		panic("blob.StoreUnderPolicy: q must not be nil")
	}
	if content == nil {
		content = []byte{}
	}
	naming, err := hash.NamingFor(q)
	if err != nil {
		return 0, "", Found, fmt.Errorf("blob: %w", err)
	}
	uuid = naming.New.Hash(content)
	outcome = Inserted
	if rid, ok := Exists(q, uuid); ok {
		if !isPhantom(q, rid) {
			return rid, uuid, Found, nil
		}
		outcome = FilledPhantom
	}
	rid, err = writeVerified(q, naming.New, uuid, content)
	if err != nil {
		return 0, "", Found, fmt.Errorf("blob.StoreUnderPolicy: %w", err)
	}
	return rid, uuid, outcome, nil
}

// writeVerified compresses content into the row for uuid, reads it back to
// check the name, and marks it unclustered.
func writeVerified(q db.Querier, alg hash.Alg, uuid string, content []byte) (libfossil.FslID, error) {
	compressed, err := Compress(content)
	if err != nil {
		return 0, fmt.Errorf("compress: %w", err)
	}
	rid, err := writeRow(q, uuid, int64(len(content)), compressed)
	if err != nil {
		return 0, err
	}

	// Verify round-trip: re-read, decompress, re-hash.
	// Matches Fossil's content_put_pk() post-write verification.
	readBack, err := Load(q, rid)
	if err != nil {
		return 0, fmt.Errorf("verify read-back: %w", err)
	}
	if got := alg.Hash(readBack); got != uuid {
		return 0, fmt.Errorf("verify: hash mismatch after round-trip: stored %s, got %s", uuid, got)
	}

	// Mark as unclustered — matches Fossil's content_put_ex (content.c:633).
	// Only new blobs reach here; an existing one returns before.
	if _, err := q.Exec("INSERT OR IGNORE INTO unclustered(rid) VALUES(?)", rid); err != nil {
		return 0, fmt.Errorf("unclustered: %w", err)
	}
	return rid, nil
}

// StoreDelta inserts a NEW row holding content delta-encoded against
// srcRid. It has no callers outside tests and is not what the commit path
// uses: deltifying an already-stored artifact rewrites an existing row
// rather than inserting one, and the decision of what to deltify is
// content.Deltify's, which holds the whole policy in one place. Kept for
// the delta-chain fixtures several packages build with it.
func StoreDelta(q db.Querier, content []byte, srcRid libfossil.FslID) (rid libfossil.FslID, uuid string, err error) {
	if q == nil {
		panic("blob.StoreDelta: q must not be nil")
	}
	if len(content) == 0 {
		panic("blob.StoreDelta: content length must be > 0")
	}
	if srcRid <= 0 {
		panic("blob.StoreDelta: srcRid must be > 0")
	}
	defer func() {
		if err == nil && rid <= 0 {
			panic("blob.StoreDelta: postcondition violated: rid <= 0 with no error")
		}
	}()

	naming, err := hash.NamingFor(q)
	if err != nil {
		return 0, "", fmt.Errorf("blob: %w", err)
	}
	uuid = naming.New.Hash(content)

	if rid, name, ok := existingName(q, naming, uuid, content); ok {
		return rid, name, nil
	}

	srcContent, err := Load(q, srcRid)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta load source: %w", err)
	}

	deltaBytes, err := delta.Create(srcContent, content)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta delta: %w", err)
	}
	compressed, err := Compress(deltaBytes)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta compress: %w", err)
	}

	rid, err = writeRow(q, uuid, int64(len(content)), compressed)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta: %w", err)
	}
	_, err = q.Exec("INSERT INTO delta(rid, srcid) VALUES(?, ?)", rid, srcRid)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta insert delta: %w", err)
	}

	// Verify round-trip: re-read delta, apply to source, re-hash.
	// Matches Fossil's content_put_pk() post-write verification.
	storedDelta, err := Load(q, rid)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta verify read-back: %w", err)
	}
	rebuilt, err := delta.Apply(srcContent, storedDelta)
	if err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta verify apply: %w", err)
	}
	if got := naming.New.Hash(rebuilt); got != uuid {
		return 0, "", fmt.Errorf("blob.StoreDelta verify: hash mismatch after round-trip: stored %s, got %s", uuid, got)
	}

	// Mark as unclustered — matches Fossil's content_put_ex (content.c:633).
	if _, err := q.Exec("INSERT OR IGNORE INTO unclustered(rid) VALUES(?)", rid); err != nil {
		return 0, "", fmt.Errorf("blob.StoreDelta unclustered: %w", err)
	}

	return rid, uuid, nil
}

// StoreDeltaRaw stores delta-encoded content exactly as given — without
// expanding it against its source — and records the delta-to-source link.
// Unlike StoreDelta, which computes a delta from full target content and
// verifies the round-trip against an already-readable source, StoreDeltaRaw
// accepts content that arrived already delta-encoded (e.g. over the wire
// during a transfer) whose source may itself still be a phantom. The
// target's size is read from the delta's own header (delta.OutputSize), so
// no source content is needed to store it.
//
// If a real (non-phantom) blob already exists for uuid, this is a no-op.
// If a phantom row exists, it is filled in place. Otherwise a new row is
// created. In every case the row's declared size is real, never -1: the
// target is not phantomized just because its source might be — mirrors
// Fossil's content_put_ex (src/content.c:557-620), which stores delta
// content unconditionally and records REPLACE INTO delta(rid,srcid)
// whether or not the source is currently available. Availability is a
// live, transitively-recomputed property (content.IsAvailable), not
// something this function or any caller needs to update or be notified
// about; content.Expand verifies the expanded result against its claimed
// UUID on every read, since nothing here can verify a delta whose source
// isn't readable yet.
//
// deltaBytes is wire data, not a programmer-controlled argument: malformed
// or empty input (including from a hostile peer) is reported as an error,
// never a panic. Argument shapes StoreDeltaRaw's own caller controls (q,
// uuid, srcRid) are still asserted, matching this package's usual
// nil/invalid-argument convention.
//
// storedBlob, when non-nil, is deltaBytes already re-expressed in Fossil's
// on-disk blob format (e.g. bytes received over the wire already encoded
// that way) and is written verbatim via EncodeForStorage instead of being
// recompressed. Pass nil when no such bytes are available (e.g. a delta
// computed locally rather than received).
func StoreDeltaRaw(q db.Querier, uuid string, deltaBytes []byte, srcRid libfossil.FslID, storedBlob []byte) (rid libfossil.FslID, err error) {
	if q == nil {
		panic("blob.StoreDeltaRaw: q must not be nil")
	}
	if uuid == "" {
		panic("blob.StoreDeltaRaw: uuid must not be empty")
	}
	if srcRid <= 0 {
		panic("blob.StoreDeltaRaw: srcRid must be > 0")
	}
	defer func() {
		if err == nil && rid <= 0 {
			panic("blob.StoreDeltaRaw: postcondition violated: rid <= 0 with no error")
		}
	}()

	if existingRid, exists := Exists(q, uuid); exists && !isPhantom(q, existingRid) {
		return existingRid, nil // real blob already exists; nothing to do
	}

	// delta.OutputSize rejects empty, malformed, and overflowing/oversized
	// headers with an error — the boundary for wire-data validation lives
	// there so this function doesn't duplicate it.
	targetSize, err := delta.OutputSize(deltaBytes)
	if err != nil {
		return 0, fmt.Errorf("blob.StoreDeltaRaw: output size: %w", err)
	}
	compressed, err := EncodeForStorage(deltaBytes, storedBlob)
	if err != nil {
		return 0, fmt.Errorf("blob.StoreDeltaRaw: compress: %w", err)
	}

	rid, err = writeRow(q, uuid, int64(targetSize), compressed)
	if err != nil {
		return 0, fmt.Errorf("blob.StoreDeltaRaw: %w", err)
	}

	// Record the delta-to-source link unconditionally — matches
	// content_put_ex's REPLACE INTO delta(rid,srcid), which fires
	// regardless of whether srcRid is itself currently available.
	if _, err := q.Exec("REPLACE INTO delta(rid, srcid) VALUES(?, ?)", rid, srcRid); err != nil {
		return 0, fmt.Errorf("blob.StoreDeltaRaw: delta link: %w", err)
	}
	if _, err := q.Exec("INSERT OR IGNORE INTO unclustered(rid) VALUES(?)", rid); err != nil {
		return 0, fmt.Errorf("blob.StoreDeltaRaw: unclustered: %w", err)
	}

	return rid, nil
}

func StorePhantom(q db.Querier, uuid string) (rid libfossil.FslID, err error) {
	if q == nil {
		panic("blob.StorePhantom: q must not be nil")
	}
	if uuid == "" {
		panic("blob.StorePhantom: uuid must not be empty")
	}
	defer func() {
		if err == nil && rid <= 0 {
			panic("blob.StorePhantom: postcondition violated: rid <= 0 with no error")
		}
	}()

	if rid, ok := Exists(q, uuid); ok {
		return rid, nil
	}

	result, err := q.Exec(
		"INSERT INTO blob(uuid, size, content, rcvid) VALUES(?, -1, NULL, 0)",
		uuid,
	)
	if err != nil {
		return 0, fmt.Errorf("blob.StorePhantom: %w", err)
	}

	ridInt, err := result.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("blob.StorePhantom lastid: %w", err)
	}

	rid = libfossil.FslID(ridInt)
	_, err = q.Exec("INSERT INTO phantom(rid) VALUES(?)", rid)
	if err != nil {
		return 0, fmt.Errorf("blob.StorePhantom phantom table: %w", err)
	}

	return rid, nil
}

func Load(q db.Querier, rid libfossil.FslID) (result []byte, err error) {
	return loadWith(q, rid, nil)
}

// loadWith is the shared body of Load and Inflater.Load. inf, when non-nil,
// decompresses through a reusable zlib reader instead of building a fresh one;
// the uncompressed fast path needs no reader and is identical either way.
func loadWith(q db.Querier, rid libfossil.FslID, inf *Inflater) (result []byte, err error) {
	if q == nil {
		panic("blob.Load: q must not be nil")
	}
	if rid <= 0 {
		panic("blob.Load: rid must be > 0")
	}
	defer func() {
		if err == nil && result == nil {
			panic("blob.Load: postcondition violated: result is nil with no error")
		}
	}()

	content, err := readStored(q, rid)
	if err != nil {
		return nil, fmt.Errorf("blob.Load: %w", err)
	}

	// blob.content is always [4-byte BE length][zlib data]; Fossil inflates
	// it unconditionally (content_of_blob, src/content.c:218-230) and never
	// consults blob.size to decide. Guessing "stored raw" from
	// len(content) == size is unsound: for a delta row, size is the
	// expanded artifact's length while the stored bytes are the compressed
	// delta, so the two can coincide by chance -- and whether they do
	// depends on the exact bytes compress/flate emits, which differs across
	// Go releases (issue #235).
	return decompress(content, inf)
}

// decompress inflates a compressed blob, reusing inf's reader when inf is
// non-nil and falling back to a one-shot Decompress otherwise.
func decompress(data []byte, inf *Inflater) ([]byte, error) {
	if inf != nil {
		return inf.decompress(data)
	}
	return Decompress(data)
}

// Exists reports whether a blob row with the given uuid is present, and
// returns its rid. It answers existence only.
//
// It returns true for a phantom — a real blob row with size = -1 and NULL
// content, standing in for an artifact we know of but have not received.
// A phantom's content cannot be read, and neither can that of a delta whose
// chain bottoms out in one. Callers that are about to read content must use
// content.AvailableByUUID instead, which is transitive over the delta chain.
//
// Mirrors Fossil's rid_from_uuid (src/xfer.c:70).
func Exists(q db.Querier, uuid string) (libfossil.FslID, bool) {
	if q == nil {
		panic("blob.Exists: q must not be nil")
	}
	if uuid == "" {
		panic("blob.Exists: uuid must not be empty")
	}
	var rid int64
	err := q.QueryRow("SELECT rid FROM blob WHERE uuid=?", uuid).Scan(&rid)
	if err != nil {
		return 0, false
	}
	return libfossil.FslID(rid), true
}