package ingest
import (
"bytes"
"errors"
"fmt"
"hash"
"hash/crc32"
"io"
"lindenii.org/go/furgit/internal/compress/zlib"
"lindenii.org/go/furgit/internal/format/packfile"
"lindenii.org/go/furgit/internal/progress"
"lindenii.org/go/furgit/object/header"
"lindenii.org/go/furgit/object/id"
"lindenii.org/go/lgo/intconv"
)
// scanBufferSize is the stream scanner's fixed input window size.
const scanBufferSize = 64 << 10
// scanner reads one pack stream,
// mirroring consumed bytes into the destination pack file
// while maintaining the running pack hash and a per-entry CRC.
//
// It implements [io.Reader] and [io.ByteReader]
// so a zlib reader can consume an entry payload through it
// without reading past the end of that compressed stream.
type scanner struct {
src io.Reader
dst io.Writer
// buf[off:n] is the unread window.
buf []byte
off int
n int
// consumed counts stream bytes consumed so far.
consumed int
// hash accumulates the pack hash over consumed bytes
// while hashing is true.
hash hash.Hash
hashing bool
// crc accumulates the CRC of the current entry
// while crcing is true.
crc uint32
crcing bool
}
// newScanner constructs one scanner mirroring src into dst,
// seeding the running hash from the already-consumed pack header.
func newScanner(src io.Reader, dst io.Writer, packHash hash.Hash) *scanner {
return &scanner{
src: src,
dst: dst,
buf: make([]byte, scanBufferSize),
consumed: packfile.HeaderLen,
hash: packHash,
hashing: true,
crc: 0,
crcing: false,
}
}
// readPackHeader reads and validates the pack header from src,
// returning the raw header and its declared object count.
func readPackHeader(src io.Reader) ([packfile.HeaderLen]byte, int, error) {
var raw [packfile.HeaderLen]byte
_, err := io.ReadFull(src, raw[:])
if err != nil {
return raw, 0, fmt.Errorf("%w: reading header: %w", ErrMalformedPack, err)
}
packHeader, err := packfile.ParseHeader(raw[:])
if err != nil {
return raw, 0, fmt.Errorf("%w: %w", ErrMalformedPack, err)
}
count, err := intconv.Uint32ToInt(packHeader.ObjectCount)
if err != nil {
return raw, 0, fmt.Errorf("%w: object count: %w", ErrMalformedPack, err)
}
return raw, count, nil
}
// Read implements [io.Reader].
func (scanner *scanner) Read(dst []byte) (int, error) {
if len(dst) == 0 {
return 0, nil
}
err := scanner.ensureAvailable()
if err != nil {
return 0, err
}
read := min(len(dst), scanner.n-scanner.off)
copy(dst, scanner.buf[scanner.off:scanner.off+read])
err = scanner.use(read)
if err != nil {
return 0, err
}
return read, nil
}
// ReadByte implements [io.ByteReader] without allocation.
func (scanner *scanner) ReadByte() (byte, error) {
err := scanner.ensureAvailable()
if err != nil {
return 0, err
}
b := scanner.buf[scanner.off]
err = scanner.use(1)
if err != nil {
return 0, err
}
return b, nil
}
// ensureAvailable makes at least one unread byte available,
// returning [io.EOF] once the source is exhausted.
func (scanner *scanner) ensureAvailable() error {
for scanner.n-scanner.off == 0 {
err := scanner.flushPrefix()
if err != nil {
return err
}
read, err := scanner.src.Read(scanner.buf[scanner.n:])
scanner.n += read
if err != nil {
if errors.Is(err, io.EOF) {
if scanner.n-scanner.off == 0 {
return io.EOF
}
return nil
}
return fmt.Errorf("object/store/packed/internal/ingest: reading pack: %w", err)
}
if read == 0 && scanner.n-scanner.off == 0 {
return io.ErrNoProgress
}
}
return nil
}
// peekHeader returns the unread window grown to at most maxLen bytes
// without consuming, tolerating an early end of stream.
func (scanner *scanner) peekHeader(maxLen int) ([]byte, error) {
maxLen = min(maxLen, len(scanner.buf))
for scanner.n-scanner.off < maxLen {
err := scanner.flushPrefix()
if err != nil {
return nil, err
}
read, err := scanner.src.Read(scanner.buf[scanner.n:])
scanner.n += read
if err != nil {
if errors.Is(err, io.EOF) {
break
}
return nil, fmt.Errorf("object/store/packed/internal/ingest: reading pack: %w", err)
}
if read == 0 {
break
}
}
if scanner.n-scanner.off == 0 {
return nil, fmt.Errorf("%w: unexpected end of stream", ErrMalformedPack)
}
return scanner.buf[scanner.off:scanner.n], nil
}
// use consumes n unread bytes,
// folding them into the running hash and entry CRC as enabled.
func (scanner *scanner) use(n int) error {
chunk := scanner.buf[scanner.off : scanner.off+n]
if scanner.hashing {
_, err := scanner.hash.Write(chunk)
if err != nil {
return fmt.Errorf("object/store/packed/internal/ingest: hashing pack: %w", err)
}
}
if scanner.crcing {
scanner.crc = crc32.Update(scanner.crc, crc32.IEEETable, chunk)
}
scanner.off += n
scanner.consumed += n
return nil
}
// flushPrefix writes the consumed buffer prefix to the destination
// and compacts the unread window to the start of the buffer.
func (scanner *scanner) flushPrefix() error {
if scanner.off == 0 {
return nil
}
_, err := scanner.dst.Write(scanner.buf[:scanner.off])
if err != nil {
return fmt.Errorf("object/store/packed/internal/ingest: writing pack: %w", err)
}
unread := scanner.n - scanner.off
copy(scanner.buf, scanner.buf[scanner.off:scanner.n])
scanner.off = 0
scanner.n = unread
return nil
}
// beginCRC starts CRC accumulation for one entry.
func (scanner *scanner) beginCRC() {
scanner.crc = 0
scanner.crcing = true
}
// endCRC ends CRC accumulation and returns the entry CRC.
func (scanner *scanner) endCRC() uint32 {
crc := scanner.crc
scanner.crc = 0
scanner.crcing = false
return crc
}
// finishTrailer reads and verifies the pack trailer hash,
// flushing the remaining buffered pack bytes to the destination.
//
// The trailer is mirrored to the destination but excluded from the pack hash.
func (scanner *scanner) finishTrailer(hashSize int) ([]byte, error) {
trailer := make([]byte, hashSize)
scanner.hashing = false
_, err := io.ReadFull(scanner, trailer)
if err != nil {
return nil, fmt.Errorf("%w: reading trailer: %w", ErrMalformedPack, err)
}
if scanner.n-scanner.off > 0 {
return nil, fmt.Errorf("%w: trailing data after pack", ErrMalformedPack)
}
if !bytes.Equal(scanner.hash.Sum(nil), trailer) {
return nil, fmt.Errorf("%w: trailer hash mismatch", ErrMalformedPack)
}
err = scanner.flushPrefix()
if err != nil {
return nil, err
}
return trailer, nil
}
// streamAndScan streams the pack body to the temporary pack file,
// scanning one record per declared object and verifying the trailer.
func (ingestion *ingestion) streamAndScan() error {
meter := progress.New(progress.Options{
Writer: ingestion.opts.Progress,
Title: "receiving objects",
Total: ingestion.headerCount,
Delay: 0,
Sparse: false,
Throughput: true,
})
for done := range ingestion.headerCount {
err := ingestion.scanEntry(ingestion.scanner.consumed)
if err != nil {
return err
}
meter.Set(done+1, ingestion.scanner.consumed)
}
meter.Stop("done")
trailer, err := ingestion.scanner.finishTrailer(ingestion.objectFormat.Size())
if err != nil {
return err
}
packHash, err := ingestion.objectFormat.FromBytes(trailer)
if err != nil {
return fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
ingestion.packHash = packHash
return nil
}
// scanEntry scans the entry beginning at start into one record.
func (ingestion *ingestion) scanEntry(start int) error {
ingestion.scanner.beginCRC()
rec, err := ingestion.scanHeader(start)
if err != nil {
return err
}
inflated, oid, err := ingestion.drainPayload(&rec)
if err != nil {
return err
}
if inflated != int64(rec.declaredSize) {
return fmt.Errorf(
"%w: entry at %d: inflated size %d differs from declared %d",
ErrMalformedPack, start, inflated, rec.declaredSize,
)
}
rec.packedLen = ingestion.scanner.consumed - start
rec.crc32 = ingestion.scanner.endCRC()
if rec.packedType.IsBase() {
rec.objectType = rec.packedType
rec.oid = oid
rec.resolved = true
} else {
ingestion.deltaCount++
}
index := len(ingestion.records)
ingestion.records = append(ingestion.records, rec)
ingestion.byOffset[rec.offset] = index
if rec.resolved {
ingestion.byOID[rec.oid] = index
}
return nil
}
// scanHeader parses and consumes the entry header at start.
func (ingestion *ingestion) scanHeader(start int) (record, error) {
var rec record
rec.offset = start
window, err := ingestion.scanner.peekHeader(packfile.MaxEntryHeaderLen(ingestion.objectFormat.Size()))
if err != nil {
return rec, err
}
entryHeader, err := packfile.ParseEntryHeader(window, ingestion.objectFormat.Size())
if err != nil {
return rec, fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, start, err)
}
declaredSize, err := intconv.Uint64ToInt(entryHeader.Size)
if err != nil {
return rec, fmt.Errorf("%w: entry at %d: declared size overflows int: %w", ErrMalformedPack, start, err)
}
rec.packedType = entryHeader.Type
rec.declaredSize = declaredSize
rec.headerLen = entryHeader.HeaderLen
switch entryHeader.Type {
case packfile.EntryTypeCommit, packfile.EntryTypeTree, packfile.EntryTypeBlob, packfile.EntryTypeTag:
case packfile.EntryTypeOfsDelta:
dist, err := intconv.Uint64ToInt(entryHeader.OfsDistance)
if err != nil || dist == 0 || dist > start {
return rec, fmt.Errorf("%w: entry at %d: ofs-delta base out of bounds", ErrMalformedPack, start)
}
rec.baseOffset = start - dist
case packfile.EntryTypeRefDelta:
baseID, err := ingestion.objectFormat.FromBytes(entryHeader.RefBase[:ingestion.objectFormat.Size()])
if err != nil {
return rec, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
rec.baseOID = baseID
case packfile.EntryTypeInvalid, packfile.EntryTypeFuture:
return rec, fmt.Errorf("%w: entry at %d: unsupported entry type", ErrMalformedPack, start)
}
err = ingestion.scanner.use(entryHeader.HeaderLen)
if err != nil {
return rec, err
}
return rec, nil
}
// drainPayload consumes one entry's compressed payload from the stream,
// returning its inflated length and, for base entries, its object ID.
func (ingestion *ingestion) drainPayload(rec *record) (int64, id.ObjectID, error) {
var zero id.ObjectID
zr, err := zlib.NewReader(ingestion.scanner)
if err != nil {
return 0, zero, fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
}
defer func() { _ = zr.Close() }()
if !rec.packedType.IsBase() {
read, err := io.Copy(io.Discard, zr)
if err != nil {
return 0, zero, fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
}
return read, zero, nil
}
objectType, err := rec.packedType.ObjectType()
if err != nil {
return 0, zero, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
hashImpl, err := ingestion.objectFormat.New()
if err != nil {
return 0, zero, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
_, _ = hashImpl.Write(header.Append(nil, objectType, rec.declaredSize))
read, err := io.Copy(hashImpl, zr)
if err != nil {
return 0, zero, fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
}
oid, err := ingestion.objectFormat.FromBytes(hashImpl.Sum(nil))
if err != nil {
return 0, zero, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
return read, oid, nil
}
