feat: implement control history and mutation APIs

internal/server/store/events_read.go

@@ -0,0 +1,156 @@
+package store
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"github.com/klauspost/compress/zstd"
+	"github.com/rvbox/rvbox/internal/domain"
+)
+
+// EventView is a validated persisted command event. Payload is the stored
+// event bytes (normally deterministic protobuf bytes); callers that need the
+// uncompressed form should use DecodeEventPayload.
+type EventView struct {
+	IssueUUID        domain.UUID
+	EventSeq         uint64
+	ObservedUnixNano int64
+	ReceiptUnixNano  int64
+	EventType        uint16
+	Stream           uint16
+	Compression      uint16
+	RawLength        uint64
+	Payload          []byte
+	ImmutableSHA256  [32]byte
+}
+
+// ReadCommandEvents returns at most limit events after afterEventSeq in strict
+// sequence order. Segment references are revalidated at read time; a malformed
+// or replaced segment is an error rather than silently returning partial data.
+func (store *Store) ReadCommandEvents(ctx context.Context, issue domain.UUID, afterEventSeq uint64, limit uint32) ([]EventView, error) {
+	if issue == (domain.UUID{}) || limit == 0 || limit > 1000 {
+		return nil, errors.New("invalid command event read")
+	}
+	database, err := store.openDatabase()
+	if err != nil {
+		return nil, err
+	}
+	rows, err := database.QueryContext(ctx, `SELECT
+ ce.event_seq, ce.observed_at, ce.server_receipt_time, ce.event_type,
+ ce.compression, ce.raw_bytes, ce.payload, ce.segment_ordinal,
+ ce.segment_record_offset, ce.segment_record_length, os.path,
+ ce.immutable_sha256
+ FROM command_events ce LEFT JOIN output_segments os
+ ON os.issue_uuid = ce.issue_uuid AND os.ordinal = ce.segment_ordinal
+ WHERE ce.issue_uuid = ? AND ce.event_seq > ? ORDER BY ce.event_seq LIMIT ?`, issue[:], afterEventSeq, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	result := make([]EventView, 0, limit)
+	for rows.Next() {
+		var event EventView
+		var inline []byte
+		var ordinal, offset, length sql.NullInt64
+		var path sql.NullString
+		var digest []byte
+		if err := rows.Scan(&event.EventSeq, &event.ObservedUnixNano, &event.ReceiptUnixNano, &event.EventType, &event.Compression, &event.RawLength, &inline, &ordinal, &offset, &length, &path, &digest); err != nil {
+			return nil, err
+		}
+		if len(digest) != 32 || event.EventSeq == 0 || event.EventType == 0 {
+			return nil, ErrInvalidSegmentRecord
+		}
+		copy(event.IssueUUID[:], issue[:])
+		copy(event.ImmutableSHA256[:], digest)
+		switch {
+		case inline != nil && !ordinal.Valid && !offset.Valid && !length.Valid && !path.Valid:
+			event.Payload = append([]byte(nil), inline...)
+		case inline == nil && ordinal.Valid && offset.Valid && length.Valid && path.Valid:
+			if ordinal.Int64 < 0 || offset.Int64 < 0 || length.Int64 <= 0 || ordinal.Int64 > int64(^uint32(0)) || uint64(length.Int64) > uint64(^uint(0)>>1) {
+				return nil, ErrInvalidSegmentRecord
+			}
+			payload, stream, readErr := store.readEventSegment(path.String, issue, event.EventSeq, uint32(ordinal.Int64), uint64(offset.Int64), uint64(length.Int64))
+			if readErr != nil {
+				return nil, readErr
+			}
+			event.Payload = payload
+			event.Stream = stream
+		default:
+			return nil, ErrInvalidSegmentRecord
+		}
+		if uint64(len(event.Payload)) > DefaultSegmentLimit || event.Compression < 1 || event.Compression > 2 || event.RawLength > DefaultSegmentLimit {
+			return nil, ErrInvalidSegmentRecord
+		}
+		result = append(result, event)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+
+func (store *Store) readEventSegment(name string, issue domain.UUID, sequence uint64, ordinal uint32, offset, length uint64) ([]byte, uint16, error) {
+	if filepath.Base(name) != name {
+		return nil, 0, ErrUnsafeSegmentReference
+	}
+	owner, parsedOrdinal, err := parseSegmentName(name)
+	if err != nil || owner != [16]byte(issue) || parsedOrdinal != ordinal {
+		return nil, 0, ErrUnsafeSegmentReference
+	}
+	path := filepath.Join(store.segmentDirectory(), name)
+	info, err := os.Lstat(path)
+	if err != nil {
+		return nil, 0, err
+	}
+	if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 || info.Mode().Perm()&0o077 != 0 {
+		return nil, 0, ErrUnsafeSegmentReference
+	}
+	if length > uint64(^uint(0)>>1) || offset > uint64(^uint(0)>>1) || offset+length < offset || offset+length > uint64(info.Size()) {
+		return nil, 0, ErrCommittedRangeMissing
+	}
+	file, err := os.Open(path)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer file.Close()
+	reader := io.NewSectionReader(file, int64(offset), int64(length))
+	record, encodedLength, err := DecodeSegmentRecord(reader, DefaultSegmentLimits())
+	if err != nil {
+		return nil, 0, fmt.Errorf("decode command event segment: %w", err)
+	}
+	if encodedLength != length || record.OwnerUUID != [16]byte(issue) || record.Sequence != sequence || record.Kind != PayloadKindCommandEvent {
+		return nil, 0, ErrInvalidSegmentRecord
+	}
+	return append([]byte(nil), record.Payload...), record.Stream, nil
+}
+
+// DecodeEventPayload validates and boundedly decompresses a stored event
+// payload. It is intentionally generic so output/follow handlers share the
+// same decompression ceiling.
+func DecodeEventPayload(event EventView) ([]byte, error) {
+	if event.Compression == 1 {
+		if event.RawLength != uint64(len(event.Payload)) {
+			return nil, ErrInvalidSegmentRecord
+		}
+		return append([]byte(nil), event.Payload...), nil
+	}
+	if event.Compression != 2 || event.RawLength > DefaultSegmentLimit {
+		return nil, ErrInvalidSegmentRecord
+	}
+	decoder, err := zstd.NewReader(bytes.NewReader(event.Payload), zstd.WithDecoderConcurrency(1), zstd.WithDecoderMaxMemory(DefaultSegmentLimit+1))
+	if err != nil {
+		return nil, ErrInvalidSegmentRecord
+	}
+	defer decoder.Close()
+	decoded, err := io.ReadAll(io.LimitReader(decoder, int64(event.RawLength)+1))
+	if err != nil || uint64(len(decoded)) != event.RawLength {
+		return nil, ErrInvalidSegmentRecord
+	}
+	return decoded, nil
+}