Add CDX records generation capabilities

.gitignore

@@ -0,0 +1,3 @@
+main.go
+gocdx
+.vscode/

generate.go

@@ -0,0 +1,280 @@
+package gocdx
+
+import (
+	"bufio"
+	"context"
+	"crypto/sha1"
+	"encoding/base32"
+	"io"
+	"runtime"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/internetarchive/gocdx/pkg/surt"
+	warc "github.com/internetarchive/gowarc"
+	"golang.org/x/sync/errgroup"
+)
+
+type overloadedWARCRecord struct {
+	*warc.Record
+
+	compByteOffset int64
+	compByteLength int64
+	warcFileName   string
+}
+
+// contains helper
+func hasField(fields []string, want string) bool {
+	for _, f := range fields {
+		if f == want {
+			return true
+		}
+	}
+	return false
+}
+
+// Generate reads a WARC file from the provided reader and returns a slice of Record
+// generated from the given WARC response records. Reading from the WARC reader is
+// strictly single-threaded; record processing is concurrent.
+func Generate(warcFile io.Reader, header string) ([]*Record, error) {
+	warcReader, err := warc.NewReader(warcFile)
+	if err != nil {
+		return nil, err
+	}
+	defer warcReader.Close()
+
+	// Which fields are requested?
+	headerFields := strings.Fields(header)
+	needStatus := hasField(headerFields, "s") // requires parsing HTTP status line
+	needDigest := hasField(headerFields, "k") // may require reading full record content
+
+	// ---- Stage 1: single-threaded collection (the only critical section) ----
+	var (
+		warcFileName string
+		warcRecords  []*overloadedWARCRecord // only "response" records are collected
+	)
+
+	for {
+		rec, err := warcReader.ReadRecord()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			// Best effort: close any content we already collected
+			for _, r := range warcRecords {
+				if r.Record != nil && r.Record.Content != nil {
+					_ = r.Record.Content.Close()
+				}
+			}
+			return nil, err
+		}
+
+		switch rec.Header.Get("WARC-Type") {
+		case "warcinfo":
+			warcFileName = rec.Header.Get("WARC-Filename")
+		case "response":
+			// Keep Content open; workers will parse and then close it later.
+			warcRecords = append(warcRecords, &overloadedWARCRecord{
+				Record:         rec,
+				compByteOffset: rec.Offset,
+				compByteLength: rec.Size,
+				warcFileName:   warcFileName,
+			})
+		default:
+		}
+	}
+
+	// Nothing to do?
+	if len(warcRecords) == 0 {
+		return []*Record{}, nil
+	}
+
+	// ---- Stage 2: concurrent processing of collected response records ----
+	records := make([]*Record, len(warcRecords))
+
+	workers := runtime.GOMAXPROCS(0) // reasonable default; tune if needed
+	g, _ := errgroup.WithContext(context.Background())
+
+	// Work distribution: each worker pulls indexes from a channel
+	type job struct{ idx int }
+	jobs := make(chan job)
+
+	// Start workers
+	for w := 0; w < workers; w++ {
+		g.Go(func() error {
+			for j := range jobs {
+				wr := warcRecords[j.idx]
+				rec := &Record{}
+
+				// Build the record according to requested fields only.
+				for _, field := range headerFields {
+					switch field {
+					case "N":
+						rec.MassagedURL = surt.Massage(wr.Record.Header.Get("WARC-Target-URI"))
+
+					case "b":
+						parsedTime, err := time.Parse(time.RFC3339, wr.Record.Header.Get("WARC-Date"))
+						if err != nil {
+							parsedTime = time.Time{}
+						}
+						rec.Timestamp = parsedTime
+
+					case "a":
+						rec.OriginalURL = wr.Record.Header.Get("WARC-Target-URI")
+
+					case "m":
+						rec.MIMEType = strings.TrimSuffix(
+							wr.Record.Header.Get("Content-Type"),
+							"; msgtype=response",
+						)
+
+					case "s":
+						// Only parse HTTP status if requested
+						rec.StatusCode = -1
+						if needStatus {
+							httpMessage, _ := parseHTTPHeadersFromWARCRecord(wr.Record)
+							parts := strings.Split(httpMessage, " ")
+							if len(parts) >= 2 {
+								if sc, err := strconv.Atoi(parts[1]); err == nil {
+									rec.StatusCode = sc
+								}
+							}
+							// Reset content for any subsequent reads (e.g., digest)
+							if rs, ok := wr.Record.Content.(io.ReadSeeker); ok {
+								_, _ = rs.Seek(0, io.SeekStart)
+							}
+						}
+
+					case "k":
+						// Prefer the digest from header if present; otherwise compute it.
+						trimmed := strings.TrimPrefix(wr.Record.Header.Get("WARC-Block-Digest"), "sha1:")
+						if trimmed != wr.Record.Header.Get("WARC-Block-Digest") {
+							rec.NewStyleChecksum = trimmed
+						} else if needDigest {
+							hasher := sha1.New()
+							// Ensure we start from the beginning (status parsing may have read some bytes)
+							if rs, ok := wr.Record.Content.(io.ReadSeeker); ok {
+								_, _ = rs.Seek(0, io.SeekStart)
+							}
+							if _, err := io.Copy(hasher, wr.Record.Content); err != nil {
+								// Close before returning error
+								_ = wr.Record.Content.Close()
+								return err
+							}
+							rec.NewStyleChecksum = base32.StdEncoding.EncodeToString(hasher.Sum(nil))
+							// Reset again is not necessary since we'll close Content below
+						}
+
+					case "r":
+						// TODO: clarify; keep placeholder
+						rec.Redirect = "-"
+
+					case "M":
+						// TODO: ignore for now
+						rec.MetaTags = "-"
+
+					case "S":
+						rec.CompressedRecordSize = wr.compByteLength
+
+					case "V":
+						rec.CompressedArcOffset = wr.compByteOffset
+
+					case "g":
+						rec.Filename = wr.warcFileName
+					}
+				}
+
+				// Release resources for this record
+				if wr.Record != nil && wr.Record.Content != nil {
+					_ = wr.Record.Content.Close()
+				}
+
+				records[j.idx] = rec
+			}
+			return nil
+		})
+	}
+
+	// Feed jobs
+	go func() {
+		for i := range warcRecords {
+			jobs <- job{idx: i}
+		}
+		close(jobs)
+	}()
+
+	// Wait for all workers
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	return records, nil
+}
+
+// FormatCDX formats a Record into a CDX string based on the header format.
+func (r *Record) FormatCDX(header string) (string, error) {
+	var result strings.Builder
+	headerFields := strings.FieldsSeq(header)
+
+	for field := range headerFields {
+		switch field {
+		case "N":
+			result.WriteString(r.MassagedURL)
+		case "b":
+			result.WriteString(r.Timestamp.Format("20060102150405"))
+		case "a":
+			result.WriteString(r.OriginalURL)
+		case "m":
+			result.WriteString(r.MIMEType)
+		case "s":
+			result.WriteString(strconv.Itoa(r.StatusCode))
+		case "k":
+			result.WriteString(r.NewStyleChecksum)
+		case "r":
+			result.WriteString(r.Redirect)
+		case "M":
+			result.WriteString(r.MetaTags)
+		case "S":
+			result.WriteString(strconv.FormatInt(r.CompressedRecordSize, 10))
+		case "V":
+			result.WriteString(strconv.FormatInt(r.CompressedArcOffset, 10))
+		case "g":
+			result.WriteString(r.Filename)
+		case "CDX":
+			continue
+		}
+		result.WriteString(" ")
+	}
+
+	return strings.TrimSpace(result.String()), nil
+}
+
+func parseHTTPHeadersFromWARCRecord(warcRecord *warc.Record) (message string, headers map[string]string) {
+	headers = make(map[string]string)
+
+	var i int
+	scanner := bufio.NewScanner(warcRecord.Content)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if line == "" {
+			break
+		}
+
+		// handle HTTP message
+		if i == 0 {
+			message = strings.Clone(line)
+			i++
+			continue
+		}
+
+		parts := strings.SplitN(line, ": ", 2)
+		if len(parts) == 2 {
+			headers[strings.ToLower(parts[0])] = strings.ToLower(parts[1])
+		}
+
+		i++
+	}
+
+	return
+}

go.mod

@@ -1,3 +1,31 @@
 module github.com/internetarchive/gocdx
 
-go 1.23.1
+go 1.24.2
+
+replace github.com/internetarchive/gowarc => ../gowarc
+
+require (
+	github.com/ImVexed/fasturl v0.0.0-20230304231329-4e41488060f3
+	github.com/grafana/pyroscope-go v1.2.4
+	github.com/internetarchive/gowarc v0.8.85
+	golang.org/x/sync v0.13.0
+)
+
+require (
+	github.com/andybalholm/brotli v1.1.1 // indirect
+	github.com/cloudflare/circl v1.6.1 // indirect
+	github.com/dolthub/maphash v0.1.0 // indirect
+	github.com/gammazero/deque v1.0.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/grafana/pyroscope-go/godeltaprof v0.1.8 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/maypok86/otter v1.2.4 // indirect
+	github.com/miekg/dns v1.1.65 // indirect
+	github.com/refraction-networking/utls v1.6.7 // indirect
+	github.com/ulikunitz/xz v0.5.12 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/mod v0.24.0 // indirect
+	golang.org/x/net v0.39.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/tools v0.32.0 // indirect
+)

go.sum

@@ -0,0 +1,62 @@
+github.com/ImVexed/fasturl v0.0.0-20230304231329-4e41488060f3 h1:ClzzXMDDuUbWfNNZqGeYq4PnYOlwlOVIvSyNaIy0ykg=
+github.com/ImVexed/fasturl v0.0.0-20230304231329-4e41488060f3/go.mod h1:we0YA5CsBbH5+/NUzC/AlMmxaDtWlXeNsqrwXjTzmzA=
+github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
+github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
+github.com/cloudflare/circl v1.6.1 h1:zqIqSPIndyBh1bjLVVDHMPpVKqp8Su/V+6MeDzzQBQ0=
+github.com/cloudflare/circl v1.6.1/go.mod h1:uddAzsPgqdMAYatqJ0lsjX1oECcQLIlRpzZh3pJrofs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dolthub/maphash v0.1.0 h1:bsQ7JsF4FkkWyrP3oCnFJgrCUAFbFf3kOl4L/QxPDyQ=
+github.com/dolthub/maphash v0.1.0/go.mod h1:gkg4Ch4CdCDu5h6PMriVLawB7koZ+5ijb9puGMV50a4=
+github.com/dvyukov/go-fuzz v0.0.0-20200318091601-be3528f3a813/go.mod h1:11Gm+ccJnvAhCNLlf5+cS9KjtbaD5I5zaZpFMsTHWTw=
+github.com/gammazero/deque v1.0.0 h1:LTmimT8H7bXkkCy6gZX7zNLtkbz4NdS2z8LZuor3j34=
+github.com/gammazero/deque v1.0.0/go.mod h1:iflpYvtGfM3U8S8j+sZEKIak3SAKYpA5/SQewgfXDKo=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/grafana/pyroscope-go v1.2.4 h1:B22GMXz+O0nWLatxLuaP7o7L9dvP0clLvIpmeEQQM0Q=
+github.com/grafana/pyroscope-go v1.2.4/go.mod h1:zzT9QXQAp2Iz2ZdS216UiV8y9uXJYQiGE1q8v1FyhqU=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.8 h1:iwOtYXeeVSAeYefJNaxDytgjKtUuKQbJqgAIjlnicKg=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.8/go.mod h1:2+l7K7twW49Ct4wFluZD3tZ6e0SjanjcUUBPVD/UuGU=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/maypok86/otter v1.2.4 h1:HhW1Pq6VdJkmWwcZZq19BlEQkHtI8xgsQzBVXJU0nfc=
+github.com/maypok86/otter v1.2.4/go.mod h1:mKLfoI7v1HOmQMwFgX4QkRk23mX6ge3RDvjdHOWG4R4=
+github.com/miekg/dns v1.1.65 h1:0+tIPHzUW0GCge7IiK3guGP57VAw7hoPDfApjkMD1Fc=
+github.com/miekg/dns v1.1.65/go.mod h1:Dzw9769uoKVaLuODMDZz9M6ynFU6Em65csPuoi8G0ck=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/refraction-networking/utls v1.6.7 h1:zVJ7sP1dJx/WtVuITug3qYUq034cDq9B2MR1K67ULZM=
+github.com/refraction-networking/utls v1.6.7/go.mod h1:BC3O4vQzye5hqpmDTWUqi4P5DDhzJfkV1tdqtawQIH0=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/things-go/go-socks5 v0.0.6 h1:YjylIYZiND41szH4NzsVbx8aVDsS/Y8ps3QYPwQvqnI=
+github.com/things-go/go-socks5 v0.0.6/go.mod h1:RF6tRutwNWzISbPfiDEChH/o1aDfRv+cXDYn2a2qkK4=
+github.com/ulikunitz/xz v0.5.12 h1:37Nm15o69RwBkXM0J6A5OlE67RZTfzUxTj8fB3dfcsc=
+github.com/ulikunitz/xz v0.5.12/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14=
+github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
+github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
+golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
+golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU=
+golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
+golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=
+golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E=
+golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
+golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
+golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/tools v0.32.0 h1:Q7N1vhpkQv7ybVzLFtTjvQya2ewbwNDZzUgfXGqtMWU=
+golang.org/x/tools v0.32.0/go.mod h1:ZxrU41P/wAbZD8EDa6dDCa6XfpkhJ7HFMjHJXfBDu8s=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=