flowstore

type Endpoint string

type FlowStore interface {
	// WriteWindow stores every record of w, folds each into the totals,
	// and stores its gaps, atomically; a gap already stored is not stored
	// again. It returns the number of records stored.
	WriteWindow(ctx context.Context, w Window) (int, error)
	// WriteGaps stores gaps the control plane derived itself for one
	// workload, outside any reported window; a gap already stored is not
	// stored again. The agent-reported gaps ride WriteWindow.
	WriteGaps(ctx context.Context, id identity.WorkloadID, gaps []Gap) error
	// ListGaps returns the gaps of a workload set intersecting a range.
	ListGaps(ctx context.Context, q GapQuery) ([]GapRow, error)
	// ListWindows returns a workload's stored windows, newest first.
	ListWindows(ctx context.Context, q WindowQuery) ([]WindowRow, error)
	// Rollup groups the windows of a workload set by peer and service.
	Rollup(ctx context.Context, q RollupQuery) ([]RollupRow, error)
	// RollupGroups aggregates windows under one of the named groupings,
	// ordered and bounded as the query says. It returns ErrUnknownGroupBy
	// for a grouping outside GroupBys.
	RollupGroups(ctx context.Context, q GroupQuery) (*GroupResult, error)
	// ListWindowPage returns one page of a workload's windows, newest
	// first, continuing after the query's cursor.
	ListWindowPage(ctx context.Context, q WindowPageQuery) ([]WindowRow, error)
	// ListTotals returns a workload's cumulative flow keys, most recently
	// seen first. A zero decision means every decision.
	ListTotals(ctx context.Context, id identity.WorkloadID, decision innerwallv1.PolicyDecision) ([]Total, error)
	// PruneWindows deletes windows that started before horizon, and gaps
	// that ended before it, in bounded batches. It reports how many window
	// rows it deleted and whether it ran at all: only one replica prunes at
	// a time, and a replica that finds the retention lock held by another
	// returns ran == false and deleted == 0.
	PruneWindows(ctx context.Context, horizon time.Time) (deleted int64, ran bool, err error)
}

type Gap struct {
	Kind   innerwallv1.EvidenceGapKind
	Source innerwallv1.EvidenceSource
	From   time.Time
	To     time.Time
	Count  *uint64
}

type GapQuery struct {
	WorkloadIDs []identity.WorkloadID
	Since       time.Time
	Until       time.Time
	Limit       int
}

type GapRow struct {
	ID         int64
	WorkloadID identity.WorkloadID
	ReceivedAt time.Time
	Gap
}

type GapWriter interface {
	WriteGaps(ctx context.Context, id identity.WorkloadID, gaps []Gap) error
}

type Group struct {
	RuleID     string
	Peer       Peer
	WorkloadID identity.WorkloadID
	DstPort    uint16
	Protocol   innerwallv1.Protocol

	// WorkloadCount is the number of distinct workloads that saw the
	// group. Only rule and peer,service count it; it is zero for the
	// others: src,dst and dst,service name their one workload in their
	// keys, and rule,peer, whose key names none, is not counted.
	WorkloadCount int64
	// FlowCount is the number of stored records in the group.
	FlowCount       int64
	ConnectionCount uint64
	ByteCount       uint64
	FirstSeen       time.Time
	LastSeen        time.Time
}

type GroupBy string

type GroupOrder string

type GroupQuery struct {
	GroupBy     GroupBy
	WorkloadIDs []identity.WorkloadID
	Endpoint    Endpoint
	Since       time.Time
	Until       time.Time
	Decision    innerwallv1.PolicyDecision
	Direction   innerwallv1.Direction
	Protocol    innerwallv1.Protocol
	DstPort     uint16
	Order       GroupOrder
	// Limit bounds the groups returned; DefaultGroupLimit when zero,
	// never more than MaxGroupLimit.
	Limit int
}

type GroupResult struct {
	Groups          []Group
	EffectiveFrom   time.Time
	EffectiveTo     time.Time
	GroupCount      int64
	FlowCount       int64
	ConnectionCount uint64
	ByteCount       uint64
}

type Peer struct {
	Kind   PeerKind
	Key    string
	Labels map[string]string
}

type PeerKind int32

type Postgres struct {
	// contains filtered or unexported fields
}

type Pruner interface {
	Prune(ctx context.Context, now time.Time) (deleted int64, err error)
}

type PrunerFunc func(ctx context.Context, now time.Time) (int64, error)

type Record struct {
	Peer            Peer
	SrcAddress      netip.Addr
	DstAddress      netip.Addr
	DstPort         uint16
	Protocol        innerwallv1.Protocol
	Direction       innerwallv1.Direction
	Decision        innerwallv1.PolicyDecision
	MatchedRuleID   string
	ConnectionCount uint64
	ByteCount       uint64
	FirstSeen       time.Time
	LastSeen        time.Time
	ProcessName     string
}

type Retention struct {
	Store FlowStore
	// Pruners run after the windows on every round, each in turn; one
	// failing does not stop the others.
	Pruners []Pruner
	// Horizon is the age beyond which windows are deleted;
	// DefaultRetention when zero.
	Horizon time.Duration
	// Interval is the period between runs; DefaultRetentionInterval when
	// zero.
	Interval time.Duration
	Log      *slog.Logger
	// Now is the clock; time.Now if nil.
	Now func() time.Time
}

type RollupQuery struct {
	WorkloadIDs []identity.WorkloadID
	Since       time.Time
	Until       time.Time
	Decision    innerwallv1.PolicyDecision
}

type RollupRow struct {
	Peer            Peer
	DstPort         uint16
	Protocol        innerwallv1.Protocol
	Decision        innerwallv1.PolicyDecision
	Workloads       int64
	ConnectionCount uint64
	ByteCount       uint64
	FirstSeen       time.Time
	LastSeen        time.Time
}

type Total struct {
	WorkloadID      identity.WorkloadID
	Peer            Peer
	DstPort         uint16
	Protocol        innerwallv1.Protocol
	Direction       innerwallv1.Direction
	Decision        innerwallv1.PolicyDecision
	MatchedRuleID   string
	FirstSeen       time.Time
	LastSeen        time.Time
	ConnectionCount uint64
	ByteCount       uint64
	WindowCount     uint64
}

type TotalsKey struct {
	PeerKind  PeerKind
	PeerKey   string
	DstPort   uint16
	Protocol  innerwallv1.Protocol
	Direction innerwallv1.Direction
	Decision  innerwallv1.PolicyDecision
}

type Window struct {
	WorkloadID identity.WorkloadID
	Start      time.Time
	End        time.Time
	Records    []Record
	Gaps       []Gap
}

type WindowCursor struct {
	WindowStart time.Time
	ID          int64
}

type WindowPageQuery struct {
	WorkloadID identity.WorkloadID
	Since      time.Time
	Until      time.Time
	Decision   innerwallv1.PolicyDecision
	Direction  innerwallv1.Direction
	PeerKey    string
	Protocol   innerwallv1.Protocol
	DstPort    uint16
	Before     *WindowCursor
	Limit      int
}

type WindowQuery struct {
	WorkloadID identity.WorkloadID
	Since      time.Time
	Until      time.Time
	Decision   innerwallv1.PolicyDecision
	Limit      int
}

type WindowRow struct {
	ID          int64
	WorkloadID  identity.WorkloadID
	WindowStart time.Time
	WindowEnd   time.Time
	Record
}