rvbox/internal/server/store/events_read.go

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
}