nmon/internal/store/groups.go

package store

import (
	"context"
	"fmt"
	"time"
)

// Group kinds and member directions.
const (
	GroupSite     = "site"
	GroupCustomer = "customer"
	GroupRole     = "role"
	GroupCustom   = "custom"

	// DirToward means "in" on the member port is traffic toward the group's
	// subject (the customer, the site); DirAway means it is traffic away.
	DirToward = "toward"
	DirAway   = "away"
)

// Group is a named set of interfaces whose traffic is summed.
type Group struct {
	ID   int64
	Name string
	Kind string
	// AlertDown opts member ports into interface_down alerts for as long as they are members.
	AlertDown bool
}

// GroupMember is one interface's membership of a group for [EffectiveFrom,
// EffectiveTo). A nil EffectiveTo means the membership is still open. History
// is never rewritten: ending a membership sets EffectiveTo, and a port that
// changes direction gets a new row.
type GroupMember struct {
	GroupID       int64
	InterfaceID   string
	Direction     string
	EffectiveFrom time.Time
	EffectiveTo   *time.Time
}

// Covers reports whether the membership is in force at t.
func (m GroupMember) Covers(t time.Time) bool {
	return !t.Before(m.EffectiveFrom) && (m.EffectiveTo == nil || t.Before(*m.EffectiveTo))
}

// Groups manages groups and their membership history.
type Groups interface {
	// CreateGroup inserts g, generating g.ID. The name must be non-empty and unique
	// (ErrConflict otherwise) and the kind one of the Group* constants (ErrInvalid).
	CreateGroup(ctx context.Context, g *Group) error
	// ListGroups returns every group by name.
	ListGroups(ctx context.Context) ([]Group, error)
	// UpdateGroup renames a group or changes its kind or alert_down flag; ErrNotFound if absent.
	UpdateGroup(ctx context.Context, g Group) error
	// DeleteGroup removes a group and its membership history. A group named by a
	// maintenance window is ErrConflict.
	DeleteGroup(ctx context.Context, id int64) error
	// AddGroupMember opens a membership. A zero EffectiveFrom means now. An
	// overlap with another membership of the same interface in the group
	// is ErrConflict; an unknown interface or direction is ErrInvalid.
	AddGroupMember(ctx context.Context, m GroupMember) error
	// EndGroupMember closes the open membership of an interface at `at` (zero
	// means now); ErrNotFound if none is open, ErrInvalid if at precedes its start.
	EndGroupMember(ctx context.Context, groupID int64, interfaceID string, at time.Time) error
	// GroupMembers returns the memberships of a group (0 means every group) in
	// force at any instant of [from, to); a zero `to` means no upper bound and a
	// zero `from` no lower bound. Ordered by group, interface, start.
	GroupMembers(ctx context.Context, groupID int64, from, to time.Time) ([]GroupMember, error)
}

func validGroupKind(k string) bool {
	switch k {
	case GroupSite, GroupCustomer, GroupRole, GroupCustom:
		return true
	}
	return false
}

func (s *bunStore) CreateGroup(ctx context.Context, g *Group) error {
	if g.Name == "" || !validGroupKind(g.Kind) {
		return fmt.Errorf("%w: a group needs a name and a kind of site, customer, role or custom", ErrInvalid)
	}
	row := &group{Name: g.Name, Kind: g.Kind, AlertDown: g.AlertDown}
	taken, err := s.db.NewSelect().Model((*group)(nil)).Where("g.name = ?", g.Name).Exists(ctx)
	if err != nil {
		return mapErr(err)
	}
	if taken {
		return fmt.Errorf("%w: group name %q is taken", ErrConflict, g.Name)
	}
	if _, err := s.db.NewInsert().Model(row).Exec(ctx); err != nil {
		return mapErr(err)
	}
	g.ID = row.ID
	return nil
}

func (s *bunStore) ListGroups(ctx context.Context) ([]Group, error) {
	var rows []group
	if err := s.db.NewSelect().Model(&rows).Order("g.name", "g.id").Scan(ctx); err != nil {
		return nil, mapErr(err)
	}
	out := make([]Group, len(rows))
	for i, r := range rows {
		out[i] = Group{ID: r.ID, Name: r.Name, Kind: r.Kind, AlertDown: r.AlertDown}
	}
	return out, nil
}

func (s *bunStore) UpdateGroup(ctx context.Context, g Group) error {
	if g.Name == "" || !validGroupKind(g.Kind) {
		return fmt.Errorf("%w: a group needs a name and a kind of site, customer, role or custom", ErrInvalid)
	}
	res, err := s.db.NewUpdate().Model((*group)(nil)).Set("name = ?", g.Name).Set("kind = ?", g.Kind).Set("alert_down = ?", g.AlertDown).
		Where("id = ?", g.ID).Exec(ctx)
	if err != nil {
		return mapErr(err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *bunStore) DeleteGroup(ctx context.Context, id int64) error {
	return s.WithTx(ctx, func(tx Store) error {
		t := tx.(*bunStore)
		if _, err := t.db.NewDelete().Model((*groupMember)(nil)).Where("group_id = ?", id).Exec(ctx); err != nil {
			return mapErr(err)
		}
		res, err := t.db.NewDelete().Model((*group)(nil)).Where("id = ?", id).Exec(ctx)
		if err != nil {
			return mapErr(err)
		}
		if n, _ := res.RowsAffected(); n == 0 {
			return ErrNotFound
		}
		return nil
	})
}

func (s *bunStore) AddGroupMember(ctx context.Context, m GroupMember) error {
	if m.Direction != DirToward && m.Direction != DirAway {
		return fmt.Errorf("%w: direction must be toward or away", ErrInvalid)
	}
	if m.EffectiveFrom.IsZero() {
		m.EffectiveFrom = s.now()
	}
	from := utcMicro(m.EffectiveFrom)
	var to *time.Time
	if m.EffectiveTo != nil {
		t := utcMicro(*m.EffectiveTo)
		if !t.After(from) {
			return fmt.Errorf("%w: effective_to must be after effective_from", ErrInvalid)
		}
		to = &t
	}
	return s.WithTx(ctx, func(tx Store) error {
		t := tx.(*bunStore)
		if _, err := t.GetGroup(ctx, m.GroupID); err != nil {
			return fmt.Errorf("%w: group does not exist", ErrInvalid)
		}
		if _, err := t.GetInterface(ctx, m.InterfaceID); err != nil {
			return fmt.Errorf("%w: interface does not exist", ErrInvalid)
		}
		if err := t.checkNoOverlap(ctx, m.GroupID, m.InterfaceID, from, to); err != nil {
			return err
		}
		row := &groupMember{GroupID: m.GroupID, InterfaceID: m.InterfaceID, Direction: m.Direction, EffectiveFrom: from, EffectiveTo: to}
		_, err := t.db.NewInsert().Model(row).Exec(ctx)
		return mapErr(err)
	})
}

// checkNoOverlap rejects a new [from, to) that intersects an existing
// membership of the same interface in the same group.
func (s *bunStore) checkNoOverlap(ctx context.Context, groupID int64, ifaceID string, from time.Time, to *time.Time) error {
	q := s.db.NewSelect().Model((*groupMember)(nil)).Where("gm.group_id = ?", groupID).Where("gm.interface_id = ?", ifaceID).
		Where("(gm.effective_to IS NULL OR gm.effective_to > ?)", from)
	if to != nil {
		q = q.Where("gm.effective_from < ?", *to)
	}
	clash, err := q.Exists(ctx)
	if err != nil {
		return mapErr(err)
	}
	if clash {
		return fmt.Errorf("%w: interface already has an overlapping membership in this group", ErrConflict)
	}
	return nil
}

func (s *bunStore) EndGroupMember(ctx context.Context, groupID int64, interfaceID string, at time.Time) error {
	if at.IsZero() {
		at = s.now()
	}
	at = utcMicro(at)
	var row groupMember
	err := s.db.NewSelect().Model(&row).Where("gm.group_id = ?", groupID).Where("gm.interface_id = ?", interfaceID).
		Where("gm.effective_to IS NULL").Scan(ctx)
	if err != nil {
		return mapErr(err)
	}
	if !at.After(row.EffectiveFrom.UTC()) {
		return fmt.Errorf("%w: a membership cannot end at or before its start", ErrInvalid)
	}
	_, err = s.db.NewUpdate().Model((*groupMember)(nil)).Set("effective_to = ?", at).
		Where("group_id = ?", groupID).Where("interface_id = ?", interfaceID).Where("effective_from = ?", row.EffectiveFrom).Exec(ctx)
	return mapErr(err)
}

func (s *bunStore) GroupMembers(ctx context.Context, groupID int64, from, to time.Time) ([]GroupMember, error) {
	var rows []groupMember
	q := s.db.NewSelect().Model(&rows)
	if groupID != 0 {
		q = q.Where("gm.group_id = ?", groupID)
	}
	if !from.IsZero() {
		q = q.Where("(gm.effective_to IS NULL OR gm.effective_to > ?)", utcMicro(from))
	}
	if !to.IsZero() {
		q = q.Where("gm.effective_from < ?", utcMicro(to))
	}
	if err := q.OrderExpr("gm.group_id, gm.interface_id, gm.effective_from").Scan(ctx); err != nil {
		return nil, mapErr(err)
	}
	out := make([]GroupMember, len(rows))
	for i, r := range rows {
		out[i] = GroupMember{GroupID: r.GroupID, InterfaceID: r.InterfaceID, Direction: r.Direction,
			EffectiveFrom: r.EffectiveFrom.UTC(), EffectiveTo: utcp(r.EffectiveTo)}
	}
	return out, nil
}