johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package infra

import (
	"cmp"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"slices"
	"strings"
	"time"

	record "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/infra"
)

const (
	IndexMaxLive = 256

	IndexMaxEnded = 32

	IndexGoneAfter = 10 * time.Minute

	IndexDescription = "index of live deployments; not a deployment"

	IndexEndDeadline = "deadline"

	IndexEndDeleted = "deleted"
)

var (
	ErrIndexFull = fmt.Errorf("infra: the index holds %d live entries; nothing is added until one is pruned", IndexMaxLive)

	ErrNotIndex = errors.New("infra: the record at rkey self is not an index of live deployments")

	recordKeyPattern = regexp.MustCompile(`^[A-Za-z0-9._:~-]{1,512}$`)
)

type IndexKey struct {
	Collection string

	Rkey string
}

type IndexEntry struct {
	Collection string

	Rkey string

	CreatedAt time.Time

	ExpiresAt time.Time

	AddedAt time.Time

	Writer string
}

func (e IndexEntry) Key() IndexKey {
	return IndexKey{Collection: e.Collection, Rkey: e.Rkey}
}

func (e IndexEntry) Expired(now time.Time) bool {
	return !e.ExpiresAt.IsZero() && !now.Before(e.ExpiresAt)
}

type IndexEnded struct {
	Collection string

	Rkey string

	EndedAt time.Time

	EndReason string
}

func (e IndexEnded) Key() IndexKey {
	return IndexKey{Collection: e.Collection, Rkey: e.Rkey}
}

type IndexDead struct {
	IndexEnded

	Gone bool
}

type IndexChange struct {
	Add *IndexEntry

	Refresh bool

	Remove *IndexEnded

	Dead []IndexDead
}

type IndexView struct {
	Live []IndexEntry

	Ended []IndexKey
}

func IndexURI(did string) string {
	return "at://" + did + "/" + record.InfraNSID + "/" + ReservedRkey
}

func validKey(key IndexKey) bool {
	if key.Collection != record.InfraNSID && key.Collection != record.EphemeralInfraNSID {
		return false
	}
	return key.Rkey != ReservedRkey && key.Rkey != "." && key.Rkey != ".." && recordKeyPattern.MatchString(key.Rkey)
}

func salvage[T any](raw any) []T {
	items, _ := raw.([]any)
	var out []T
	for _, item := range items {
		b, err := json.Marshal(item)
		if err != nil {
			continue
		}
		var v T
		if json.Unmarshal(b, &v) == nil {
			out = append(out, v)
		}
	}
	return out
}

func parseIndex(raw json.RawMessage) (*record.Infra, bool, error) {
	var value map[string]any
	if err := json.Unmarshal(raw, &value); err != nil || value == nil {
		return nil, false, fmt.Errorf("%w: the index record is not a JSON object", ErrNotIndex)
	}
	r, err := record.ParseInfra(value)
	switch {
	case err == nil && r.Index != nil:
		return r, false, nil
	case err == nil:
		return nil, false, ErrNotIndex
	}
	index, ok := value["index"].(map[string]any)
	if typ, _ := value["$type"].(string); !ok || (typ != "" && typ != record.InfraNSID) {
		return nil, false, fmt.Errorf("%w: %v", ErrNotIndex, err)
	}
	created, _ := value["createdAt"].(string)
	if _, perr := time.Parse(time.RFC3339, created); perr != nil {
		created = ""
	}
	return &record.Infra{Type: record.InfraNSID, CreatedAt: created, Description: IndexDescription, Index: &record.InfraIndex{
		Live: salvage[record.InfraIndexEntry](index["live"]), RecentEnded: salvage[record.InfraIndexEnded](index["recentEnded"])}}, true, nil
}

func entryOf(e record.InfraIndexEntry) IndexEntry {
	out := IndexEntry{Collection: strings.TrimSpace(e.Collection), Rkey: strings.TrimSpace(e.Rkey), Writer: e.Writer}
	out.CreatedAt, _ = time.Parse(time.RFC3339, e.CreatedAt)
	out.ExpiresAt, _ = time.Parse(time.RFC3339, e.ExpiresAt)
	out.AddedAt, _ = time.Parse(time.RFC3339, e.AddedAt)
	return out
}

func endedKey(e record.InfraIndexEnded) IndexKey {
	return IndexKey{Collection: strings.TrimSpace(e.Collection), Rkey: strings.TrimSpace(e.Rkey)}
}

func DecodeIndexView(raw json.RawMessage) (IndexView, error) {
	r, _, err := parseIndex(raw)
	if err != nil {
		return IndexView{}, fmt.Errorf("%w: %v", ErrInvalidDeclaration, err)
	}
	var out IndexView
	for _, e := range r.Index.Live {
		if entry := entryOf(e); validKey(entry.Key()) {
			out.Live = append(out.Live, entry)
		}
	}
	for _, e := range r.Index.RecentEnded {
		out.Ended = append(out.Ended, endedKey(e))
	}
	return out, nil
}

func DecodeIndex(raw json.RawMessage) ([]IndexEntry, error) {
	view, err := DecodeIndexView(raw)
	return view.Live, err
}

func IndexHolds(entries []IndexEntry, key IndexKey) bool {
	return slices.ContainsFunc(entries, func(e IndexEntry) bool { return e.Key() == key })
}

func sameEntry(have IndexEntry, want IndexEntry) bool {
	return have.ExpiresAt.Truncate(time.Second).Equal(want.ExpiresAt.Truncate(time.Second)) &&
		(want.CreatedAt.IsZero() || have.CreatedAt.Truncate(time.Second).Equal(want.CreatedAt.Truncate(time.Second)))
}

func (v IndexView) Settled(ch IndexChange) bool {
	switch {
	case ch.Add != nil:
		i := slices.IndexFunc(v.Live, func(e IndexEntry) bool { return e.Key() == ch.Add.Key() })
		return i >= 0 && (!ch.Refresh || sameEntry(v.Live[i], *ch.Add)) && !slices.Contains(v.Ended, ch.Add.Key())
	case ch.Remove != nil:
		return !IndexHolds(v.Live, ch.Remove.Key())
	}
	return true
}

func stamp(t time.Time) string {
	if t.IsZero() {
		return ""
	}
	return t.UTC().Format(time.RFC3339)
}

func endedRecord(e IndexEnded, now time.Time) record.InfraIndexEnded {
	return record.InfraIndexEnded{Collection: e.Collection, Rkey: e.Rkey, EndedAt: stamp(cmp.Or(e.EndedAt, now)), EndReason: e.EndReason}
}

func ApplyIndex(raw json.RawMessage, ch IndexChange, now time.Time) (json.RawMessage, bool, error) {
	now = now.UTC()
	var r record.Infra
	changed := false
	if raw == nil {
		r = record.Infra{Type: record.InfraNSID, CreatedAt: stamp(now), Description: IndexDescription, Index: &record.InfraIndex{}}
	} else {
		parsed, malformed, err := parseIndex(raw)
		if err != nil {
			return nil, false, err
		}
		r, changed = *parsed, malformed
		if r.CreatedAt == "" {
			r.CreatedAt, changed = stamp(now), true
		}
	}
	dead := map[IndexKey]IndexDead{}
	for _, d := range ch.Dead {
		dead[d.Key()] = d
	}
	var ended []record.InfraIndexEnded
	var live []record.InfraIndexEntry
	for _, e := range r.Index.Live {
		entry := entryOf(e)
		key := entry.Key()
		d, isDead := dead[key]
		switch {
		case !validKey(key) || entry.AddedAt.IsZero() || entry.CreatedAt.IsZero():
		case ch.Remove != nil && key == ch.Remove.Key():
			ended = append(ended, endedRecord(*ch.Remove, now))
		case ch.Add != nil && key == ch.Add.Key():
			if ch.Refresh && !sameEntry(entry, *ch.Add) {
				e.ExpiresAt = stamp(ch.Add.ExpiresAt)
				if !ch.Add.CreatedAt.IsZero() {
					e.CreatedAt = stamp(ch.Add.CreatedAt)
				}
				changed = true
			}
			live = append(live, e)
			continue
		case entry.Expired(now):
			ended = append(ended, endedRecord(IndexEnded{Collection: key.Collection, Rkey: key.Rkey, EndedAt: entry.ExpiresAt, EndReason: IndexEndDeadline}, now))
		case isDead && !d.Gone:
			ended = append(ended, endedRecord(d.IndexEnded, now))
		case isDead && now.Sub(entry.AddedAt) >= IndexGoneAfter:
			ended = append(ended, endedRecord(IndexEnded{Collection: key.Collection, Rkey: key.Rkey, EndedAt: cmp.Or(d.EndedAt, now),
				EndReason: IndexEndDeleted}, now))
		default:
			live = append(live, e)
			continue
		}
		changed = true
	}
	var full error
	if ch.Add != nil && !slices.ContainsFunc(live, func(e record.InfraIndexEntry) bool { return entryOf(e).Key() == ch.Add.Key() }) {
		if len(live) >= IndexMaxLive {
			full = ErrIndexFull
		} else {
			add := *ch.Add
			live = append(live, record.InfraIndexEntry{Collection: add.Collection, Rkey: add.Rkey, CreatedAt: stamp(cmp.Or(add.CreatedAt, now)),
				ExpiresAt: stamp(add.ExpiresAt), AddedAt: stamp(now), Writer: add.Writer})
			changed = true
		}
	}
	slices.SortStableFunc(ended, func(a, b record.InfraIndexEnded) int { return strings.Compare(b.EndedAt, a.EndedAt) })
	recent := ended
	for _, e := range r.Index.RecentEnded {
		key := endedKey(e)
		switch {
		case slices.ContainsFunc(ended, func(n record.InfraIndexEnded) bool { return endedKey(n) == key }):
		case full == nil && ch.Add != nil && key == ch.Add.Key():
			changed = true
		default:
			recent = append(recent, e)
		}
	}
	if len(recent) > IndexMaxEnded {
		recent, changed = recent[:IndexMaxEnded], true
	}
	if !changed {
		return raw, false, full
	}
	index := *r.Index
	index.Live, index.RecentEnded, index.UpdatedAt = live, recent, stamp(now)
	r.Index = &index
	out, err := json.Marshal(r)
	if err != nil {
		return nil, false, fmt.Errorf("infra: encoding the index: %w", err)
	}
	return out, true, full
}