[breaking-change] Implement audit record retrieval.

This is only a breaking change if you've enabled the `audit` feature.
All past audit reports should be removed once this commit is deployed,
as both the Protobuf schema and the Snowflake epoch have changed.

flake.nix

@@ -43,7 +43,7 @@
             "-s -w"
           ];
 
-          vendorHash = "sha256-opS3f4GDczDRp7mrBzvQtK13Qi4snanX4I64FHTh7Pw=";
+          vendorHash = "sha256-LkHC/gFiSfYz9Z4bYMq1QNdapPYp8h1DSMRfFU9f7mw=";
         };
       in
       {

go.mod

@@ -12,8 +12,8 @@ require (
 	github.com/getsentry/sentry-go/slog v0.40.0
 	github.com/go-git/go-billy/v6 v6.0.0-20251126203821-7f9c95185ee0
 	github.com/go-git/go-git/v6 v6.0.0-20251128074608-48f817f57805
-	github.com/influxdata/influxdb v1.12.2
 	github.com/jpillora/backoff v1.0.0
+	github.com/kankanreno/go-snowflake v1.2.0
 	github.com/klauspost/compress v1.18.1
 	github.com/maypok86/otter/v2 v2.2.1
 	github.com/minio/minio-go/v7 v7.0.97

go.sum

@@ -57,10 +57,10 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 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/influxdata/influxdb v1.12.2 h1:Y0ZBu47gYVbDCRPMFOrlRRZ3grdqPGIJxerFysVSq+g=
-github.com/influxdata/influxdb v1.12.2/go.mod h1:EwqFMB6GKV0Huug82Msa5f8QfXhqETUmC4L9A0QZJQM=
 github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
+github.com/kankanreno/go-snowflake v1.2.0 h1:Zx2SctsH5pivIj9vyhwyDyQS23jcDJx4iT49Bjv81kk=
+github.com/kankanreno/go-snowflake v1.2.0/go.mod h1:6CZ+10PeVsFXKZUTYyJzPiRIjn1IXbInaWLCX/LDJ0g=
 github.com/kevinburke/ssh_config v1.4.0 h1:6xxtP5bZ2E4NF5tuQulISpTO2z8XbtH8cg1PWkxoFkQ=
 github.com/kevinburke/ssh_config v1.4.0/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M=
 github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=

src/audit.go

@@ -1,18 +1,61 @@
 package git_pages
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
-	"github.com/influxdata/influxdb/pkg/snowflake"
 	exponential "github.com/jpillora/backoff"
+	"github.com/kankanreno/go-snowflake"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"google.golang.org/protobuf/proto"
 	timestamppb "google.golang.org/protobuf/types/known/timestamppb"
 )
 
+var (
+	auditNotifyOkCount = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "git_pages_audit_notify_ok",
+		Help: "Count of successful audit notifications",
+	})
+	auditNotifyErrorCount = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "git_pages_audit_notify_error",
+		Help: "Count of failed audit notifications",
+	})
+)
+
+type AuditID int64
+
+func GenerateAuditID() AuditID {
+	inner, err := snowflake.NextID()
+	if err != nil {
+		panic(err)
+	}
+	return AuditID(inner)
+}
+
+func ParseAuditID(repr string) (AuditID, error) {
+	inner, err := strconv.ParseInt(repr, 16, 64)
+	if err != nil {
+		return AuditID(0), err
+	}
+	return AuditID(inner), nil
+}
+
+func (id AuditID) String() string {
+	return fmt.Sprintf("%016x", int64(id))
+}
+
+func (id AuditID) CompareTime(when time.Time) int {
+	idMillis := int64(id) >> (snowflake.MachineIDLength + snowflake.SequenceLength)
+	whenMillis := when.UTC().UnixNano() / 1e6
+	return cmp.Compare(idMillis, whenMillis)
+}
+
 func EncodeAuditRecord(auditRecord *AuditRecord) (data []byte) {
 	data, err := proto.MarshalOptions{Deterministic: true}.Marshal(auditRecord)
 	if err != nil {
@@ -29,15 +72,13 @@ func DecodeAuditRecord(data []byte) (auditRecord *AuditRecord, err error) {
 
 type auditedBackend struct {
 	Backend
-	ids *snowflake.Generator
 }
 
 var _ Backend = (*auditedBackend)(nil)
 
 func NewAuditedBackend(backend Backend) Backend {
 	if config.Feature("audit") {
-		ids := snowflake.New(config.Audit.NodeID)
-		return &auditedBackend{backend, ids}
+		return &auditedBackend{backend}
 	} else {
 		return backend
 	}
@@ -50,11 +91,12 @@ func NewAuditedBackend(backend Backend) Backend {
 // to be a 100% accurate reflection of performed actions. When in doubt, the audit records
 // should be examined together with the application logs.
 func (audited *auditedBackend) appendNewAuditRecord(ctx context.Context, record *AuditRecord) (err error) {
-	record.Timestamp = timestamppb.Now()
-
 	if config.Audit.Collect {
-		id := fmt.Sprintf("%016x", audited.ids.Next())
-		err = audited.Backend.AppendAuditRecord(ctx, id, record)
+		id := GenerateAuditID()
+		record.Id = proto.Int64(int64(id))
+		record.Timestamp = timestamppb.Now()
+
+		err = audited.Backend.AppendAuditLog(ctx, id, record)
 		if err != nil {
 			err = fmt.Errorf("audit: %w", err)
 		} else {
@@ -64,7 +106,7 @@ func (audited *auditedBackend) appendNewAuditRecord(ctx context.Context, record
 			} else {
 				subject = fmt.Sprintf("%s/%s", *record.Domain, *record.Project)
 			}
-			logc.Printf(ctx, "audit %s ok: %s %s\n", subject, record.Event.String(), id)
+			logc.Printf(ctx, "audit %s ok: %s %s\n", subject, id, record.Event.String())
 
 			// Send a notification to the audit server, if configured, and try to make sure
 			// it is delivered by retrying with exponential backoff on errors.
@@ -74,10 +116,11 @@ func (audited *auditedBackend) appendNewAuditRecord(ctx context.Context, record
 	return
 }
 
-func notifyAudit(ctx context.Context, id string) {
+func notifyAudit(ctx context.Context, id AuditID) {
 	if config.Audit.NotifyURL != nil {
 		notifyURL := config.Audit.NotifyURL.URL
-		notifyURL.RawQuery = id
+		notifyURL.RawQuery = id.String()
+
 		go func() {
 			backoff := exponential.Backoff{
 				Jitter: true,
@@ -89,9 +132,11 @@ func notifyAudit(ctx context.Context, id string) {
 				if err != nil {
 					sleepFor := backoff.Duration()
 					logc.Printf(ctx, "audit notify %s err: %s (retry in %s)", id, err, sleepFor)
+					auditNotifyErrorCount.Inc()
 					time.Sleep(sleepFor)
 				} else {
 					logc.Printf(ctx, "audit notify %s ok", id)
+					auditNotifyOkCount.Inc()
 					break
 				}
 			}

src/backend.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"iter"
 	"slices"
 	"strings"
 	"time"
@@ -29,9 +30,27 @@ const (
 )
 
 type GetManifestOptions struct {
+	// If true and the manifest is past the cache `MaxAge`, `GetManifest` blocks and returns
+	// a fresh object instead of revalidating in background and returning a stale object.
 	BypassCache bool
 }
 
+type QueryAuditLogOptions struct {
+	// Inclusive lower bound on returned audit records, per their Snowflake ID (which may differ
+	// slightly from the embedded timestamp). If zero, audit records are returned since beginning
+	// of time.
+	Since time.Time
+	// Inclusive upper bound on returned audit records, per their Snowflake ID (which may differ
+	// slightly from the embedded timestamp). If zero, audit records are returned until the end
+	// of time.
+	Until time.Time
+}
+
+type QueryAuditLogResult struct {
+	ID  AuditID
+	Err error
+}
+
 type Backend interface {
 	// Returns true if the feature has been enabled for this store, false otherwise.
 	HasFeature(ctx context.Context, feature BackendFeature) bool
@@ -82,8 +101,14 @@ type Backend interface {
 	// is discovered serving abusive content.
 	FreezeDomain(ctx context.Context, domain string, freeze bool) error
 
-	// Append an audit record to the log.
-	AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) error
+	// Append a record to the audit log.
+	AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) error
+
+	// Retrieve a single record from the audit log.
+	QueryAuditLog(ctx context.Context, id AuditID) (record *AuditRecord, err error)
+
+	// Retrieve records from the audit log by time range.
+	SearchAuditLog(ctx context.Context, opts QueryAuditLogOptions) iter.Seq[QueryAuditLogResult]
 }
 
 func CreateBackend(config *StorageConfig) (backend Backend, err error) {

src/backend_fs.go

@@ -6,7 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"io/fs"
+	iofs "io/fs"
+	"iter"
 	"os"
 	"path/filepath"
 	"strings"
@@ -154,18 +155,19 @@ func (fs *FSBackend) DeleteBlob(ctx context.Context, name string) error {
 	return fs.blobRoot.Remove(blobPath)
 }
 
-func (b *FSBackend) ListManifests(ctx context.Context) (manifests []string, err error) {
-	err = fs.WalkDir(b.siteRoot.FS(), ".", func(path string, d fs.DirEntry, err error) error {
-		if strings.Count(path, "/") > 1 {
-			return fs.SkipDir
-		}
-		_, project, _ := strings.Cut(path, "/")
-		if project == "" || strings.HasPrefix(project, ".") && project != ".index" {
+func (fs *FSBackend) ListManifests(ctx context.Context) (manifests []string, err error) {
+	err = iofs.WalkDir(fs.siteRoot.FS(), ".",
+		func(path string, entry iofs.DirEntry, err error) error {
+			if strings.Count(path, "/") > 1 {
+				return iofs.SkipDir
+			}
+			_, project, _ := strings.Cut(path, "/")
+			if project == "" || strings.HasPrefix(project, ".") && project != ".index" {
+				return nil
+			}
+			manifests = append(manifests, path)
 			return nil
-		}
-		manifests = append(manifests, path)
-		return nil
-	})
+		})
 	return
 }
 
@@ -293,10 +295,50 @@ func (fs *FSBackend) FreezeDomain(ctx context.Context, domain string, freeze boo
 	}
 }
 
-func (fs *FSBackend) AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) error {
-	if _, err := fs.auditRoot.Stat(id); err == nil {
+func (fs *FSBackend) AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) error {
+	if _, err := fs.auditRoot.Stat(id.String()); err == nil {
 		panic(fmt.Errorf("audit ID collision: %s", id))
 	}
 
-	return fs.auditRoot.WriteFile(id, EncodeAuditRecord(record), 0o644)
+	return fs.auditRoot.WriteFile(id.String(), EncodeAuditRecord(record), 0o644)
+}
+
+func (fs *FSBackend) QueryAuditLog(ctx context.Context, id AuditID) (*AuditRecord, error) {
+	if data, err := fs.auditRoot.ReadFile(id.String()); err != nil {
+		return nil, fmt.Errorf("read: %w", err)
+	} else if record, err := DecodeAuditRecord(data); err != nil {
+		return nil, fmt.Errorf("decode: %w", err)
+	} else {
+		return record, nil
+	}
+}
+
+func (fs *FSBackend) SearchAuditLog(
+	ctx context.Context, opts QueryAuditLogOptions,
+) iter.Seq[QueryAuditLogResult] {
+	return func(yield func(QueryAuditLogResult) bool) {
+		iofs.WalkDir(fs.auditRoot.FS(), ".",
+			func(path string, entry iofs.DirEntry, err error) error {
+				if path == "." {
+					return nil
+				}
+				var result QueryAuditLogResult
+				if err != nil {
+					result.Err = err
+				} else if id, err := ParseAuditID(path); err != nil {
+					result.Err = err
+				} else if !opts.Since.IsZero() && id.CompareTime(opts.Since) < 0 {
+					return nil
+				} else if !opts.Until.IsZero() && id.CompareTime(opts.Until) > 0 {
+					return nil
+				} else {
+					result.ID = id
+				}
+				if !yield(result) {
+					return iofs.SkipAll
+				} else {
+					return nil
+				}
+			})
+	}
 }

src/backend_s3.go

@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"io"
+	"iter"
 	"net/http"
 	"path"
 	"strings"
@@ -631,11 +632,13 @@ func (s3 *S3Backend) FreezeDomain(ctx context.Context, domain string, freeze boo
 	}
 }
 
-func auditObjectName(id string) string {
+func auditObjectName(id AuditID) string {
 	return fmt.Sprintf("audit/%s", id)
 }
 
-func (s3 *S3Backend) AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) error {
+func (s3 *S3Backend) AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) error {
+	logc.Printf(ctx, "s3: append audit %s\n", id)
+
 	name := auditObjectName(id)
 	data := EncodeAuditRecord(record)
 
@@ -648,3 +651,49 @@ func (s3 *S3Backend) AppendAuditRecord(ctx context.Context, id string, record *A
 	}
 	return err
 }
+
+func (s3 *S3Backend) QueryAuditLog(ctx context.Context, id AuditID) (*AuditRecord, error) {
+	logc.Printf(ctx, "s3: read audit %s\n", id)
+
+	object, err := s3.client.GetObject(ctx, s3.bucket, auditObjectName(id),
+		minio.GetObjectOptions{})
+	if err != nil {
+		return nil, err
+	}
+	defer object.Close()
+
+	data, err := io.ReadAll(object)
+	if err != nil {
+		return nil, err
+	}
+
+	return DecodeAuditRecord(data)
+}
+
+func (s3 *S3Backend) SearchAuditLog(
+	ctx context.Context, opts QueryAuditLogOptions,
+) iter.Seq[QueryAuditLogResult] {
+	return func(yield func(QueryAuditLogResult) bool) {
+		logc.Printf(ctx, "s3: query audit\n")
+
+		ctx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		prefix := "audit/"
+		for object := range s3.client.ListObjectsIter(ctx, s3.bucket, minio.ListObjectsOptions{
+			Prefix: prefix,
+		}) {
+			var result QueryAuditLogResult
+			if object.Err != nil {
+				result.Err = object.Err
+			} else if id, err := ParseAuditID(strings.TrimPrefix(object.Key, prefix)); err != nil {
+				result.Err = err
+			} else {
+				result.ID = id
+			}
+			if !yield(result) {
+				break
+			}
+		}
+	}
+}

src/config.go

@@ -148,7 +148,7 @@ type LimitsConfig struct {
 }
 
 type AuditConfig struct {
-	// Globally unique node identifier (0 to 1023 inclusive).
+	// Globally unique machine identifier (0 to 63 inclusive).
 	NodeID int `toml:"node-id"`
 	// Whether audit reports should be stored whenever an audit event occurs.
 	Collect bool `toml:"collect"`

src/main.go

@@ -16,9 +16,11 @@ import (
 	"os"
 	"runtime/debug"
 	"strings"
+	"time"
 
 	automemlimit "github.com/KimMachineGun/automemlimit/memlimit"
 	"github.com/c2h5oh/datasize"
+	"github.com/kankanreno/go-snowflake"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
@@ -85,6 +87,13 @@ func configureFallback(_ context.Context) (err error) {
 	return
 }
 
+// Thread-unsafe, must be called only during initial configuration.
+func configureAudit(_ context.Context) (err error) {
+	snowflake.SetStartTime(time.Date(2025, 12, 1, 0, 0, 0, 0, time.UTC))
+	snowflake.SetMachineID(config.Audit.NodeID)
+	return
+}
+
 func listen(ctx context.Context, name string, listen string) net.Listener {
 	if listen == "-" {
 		return nil
@@ -256,6 +265,7 @@ func Main() {
 		configureMemLimit(ctx),
 		configureWildcards(ctx),
 		configureFallback(ctx),
+		configureAudit(ctx),
 	); err != nil {
 		logc.Fatalln(ctx, err)
 	}

src/observe.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"iter"
 	"log"
 	"log/slog"
 	"math/rand/v2"
@@ -437,9 +438,33 @@ func (backend *observedBackend) FreezeDomain(ctx context.Context, domain string,
 	return
 }
 
-func (backend *observedBackend) AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) (err error) {
-	span, ctx := ObserveFunction(ctx, "AppendAudit", "audit.id", id)
-	err = backend.inner.AppendAuditRecord(ctx, id, record)
+func (backend *observedBackend) AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) (err error) {
+	span, ctx := ObserveFunction(ctx, "AppendAuditLog", "audit.id", id)
+	err = backend.inner.AppendAuditLog(ctx, id, record)
 	span.Finish()
 	return
 }
+
+func (backend *observedBackend) QueryAuditLog(ctx context.Context, id AuditID) (record *AuditRecord, err error) {
+	span, ctx := ObserveFunction(ctx, "QueryAuditLog", "audit.id", id)
+	record, err = backend.inner.QueryAuditLog(ctx, id)
+	span.Finish()
+	return
+}
+
+func (backend *observedBackend) SearchAuditLog(
+	ctx context.Context, opts QueryAuditLogOptions,
+) iter.Seq[QueryAuditLogResult] {
+	return func(yield func(QueryAuditLogResult) bool) {
+		span, ctx := ObserveFunction(ctx, "SearchAuditLog",
+			"audit.search.since", opts.Since,
+			"audit.search.until", opts.Until,
+		)
+		for result := range backend.inner.SearchAuditLog(ctx, opts) {
+			if !yield(result) {
+				break
+			}
+		}
+		span.Finish()
+	}
+}

src/schema.pb.go

@@ -654,8 +654,9 @@ func (x *Manifest) GetProblems() []*Problem {
 type AuditRecord struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// Audit event metadata.
-	Event     *AuditEvent            `protobuf:"varint,1,opt,name=event,enum=AuditEvent" json:"event,omitempty"`
+	Id        *int64                 `protobuf:"varint,1,opt,name=id" json:"id,omitempty"`
 	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp" json:"timestamp,omitempty"`
+	Event     *AuditEvent            `protobuf:"varint,3,opt,name=event,enum=AuditEvent" json:"event,omitempty"`
 	// Affected resource.
 	Domain  *string `protobuf:"bytes,10,opt,name=domain" json:"domain,omitempty"`
 	Project *string `protobuf:"bytes,11,opt,name=project" json:"project,omitempty"` // only for `*Manifest` events
@@ -695,11 +696,11 @@ func (*AuditRecord) Descriptor() ([]byte, []int) {
 	return file_schema_proto_rawDescGZIP(), []int{6}
 }
 
-func (x *AuditRecord) GetEvent() AuditEvent {
-	if x != nil && x.Event != nil {
-		return *x.Event
+func (x *AuditRecord) GetId() int64 {
+	if x != nil && x.Id != nil {
+		return *x.Id
 	}
-	return AuditEvent_InvalidEvent
+	return 0
 }
 
 func (x *AuditRecord) GetTimestamp() *timestamppb.Timestamp {
@@ -709,6 +710,13 @@ func (x *AuditRecord) GetTimestamp() *timestamppb.Timestamp {
 	return nil
 }
 
+func (x *AuditRecord) GetEvent() AuditEvent {
+	if x != nil && x.Event != nil {
+		return *x.Event
+	}
+	return AuditEvent_InvalidEvent
+}
+
 func (x *AuditRecord) GetDomain() string {
 	if x != nil && x.Domain != nil {
 		return *x.Domain
@@ -775,10 +783,11 @@ const file_schema_proto_rawDesc = "" +
 	"\bproblems\x18\a \x03(\v2\b.ProblemR\bproblems\x1aC\n" +
 	"\rContentsEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x1c\n" +
-	"\x05value\x18\x02 \x01(\v2\x06.EntryR\x05value:\x028\x01\"\xc3\x01\n" +
-	"\vAuditRecord\x12!\n" +
-	"\x05event\x18\x01 \x01(\x0e2\v.AuditEventR\x05event\x128\n" +
-	"\ttimestamp\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\ttimestamp\x12\x16\n" +
+	"\x05value\x18\x02 \x01(\v2\x06.EntryR\x05value:\x028\x01\"\xd3\x01\n" +
+	"\vAuditRecord\x12\x0e\n" +
+	"\x02id\x18\x01 \x01(\x03R\x02id\x128\n" +
+	"\ttimestamp\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\ttimestamp\x12!\n" +
+	"\x05event\x18\x03 \x01(\x0e2\v.AuditEventR\x05event\x12\x16\n" +
 	"\x06domain\x18\n" +
 	" \x01(\tR\x06domain\x12\x18\n" +
 	"\aproject\x18\v \x01(\tR\aproject\x12%\n" +
@@ -837,8 +846,8 @@ var file_schema_proto_depIdxs = []int32{
 	4,  // 4: Manifest.redirects:type_name -> RedirectRule
 	6,  // 5: Manifest.headers:type_name -> HeaderRule
 	7,  // 6: Manifest.problems:type_name -> Problem
-	2,  // 7: AuditRecord.event:type_name -> AuditEvent
-	11, // 8: AuditRecord.timestamp:type_name -> google.protobuf.Timestamp
+	11, // 7: AuditRecord.timestamp:type_name -> google.protobuf.Timestamp
+	2,  // 8: AuditRecord.event:type_name -> AuditEvent
 	8,  // 9: AuditRecord.manifest:type_name -> Manifest
 	3,  // 10: Manifest.ContentsEntry.value:type_name -> Entry
 	11, // [11:11] is the sub-list for method output_type

src/schema.proto

@@ -116,8 +116,9 @@ enum AuditEvent {
 
 message AuditRecord {
 	// Audit event metadata.
-	AuditEvent event = 1;
+	int64 id = 1;
 	google.protobuf.Timestamp timestamp = 2;
+	AuditEvent event = 3;
 
 	// Affected resource.
 	string domain = 10;