lazysecrets/internal/listing/listing.go

// Package listing lists what lazysecrets can find, for lazysecrets list:
// what Go to searches, without the TUI. What's cached comes first, at
// once; then lists that are missing or older than MaxAge load in the
// background, a few at a time, and what's new in them follows as they
// arrive. Repos are always read, from disk. Values are never listed.
package listing

import (
	"cmp"
	"context"
	"fmt"
	"slices"
	"strings"
	"time"

	"git.kebcloud.no/keb/lazysecrets/internal/cachefile"
	"git.kebcloud.no/keb/lazysecrets/internal/provider"
	"git.kebcloud.no/keb/lazysecrets/internal/ref"
)

// Workers is how many lists load at once, as in Go to.
const Workers = 4

// Entry is a source, vault or item, by its lazysecrets:// link.
type Entry struct {
	Link string `json:"link"`
	// Kind is source (a subscription, workspace or repo), vault (a Key
	// Vault, secret scope or repo file), secret, key or certificate.
	Kind string `json:"kind"`
	// Provider is azure, databricks or repo, as in links.
	Provider string `json:"provider"`
	// Path names it: its source's, vault's and own names.
	Path []string `json:"path"`
	// Account is the az account it's seen through, when there are several.
	Account string `json:"account,omitempty"`
}

type Options struct {
	Providers []provider.Provider
	// Cache is what was loaded before; Run adds what it loads.
	Cache cachefile.File
	// MaxAge: cached lists loaded longer ago are loaded again.
	MaxAge time.Duration
	// Cached lists only what's cached (and repos): nothing is loaded.
	Cached  bool
	Timeout func(provider.Kind) time.Duration
	Now     func() time.Time
}

// Run lists everything, calling emit for each entry once, from this
// goroutine. An error from emit (the reader went away) stops it. It
// returns the cache with what was loaded, and the lists that couldn't be.
func Run(ctx context.Context, o Options, emit func(Entry) error) (cachefile.File, []error) {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	l := &lister{o: o, emit: emit, seen: map[string]bool{}, pending: map[string]bool{}, results: make(chan result)}
	if l.o.Now == nil {
		l.o.Now = time.Now
	}
	for _, p := range o.Providers {
		if l.stopped != nil {
			break
		}
		l.provider(p)
	}
	l.start(ctx)
	for l.active > 0 {
		r := <-l.results
		l.active--
		if l.stopped == nil {
			l.handle(r)
			l.start(ctx)
		}
		if l.stopped != nil {
			cancel()
		}
	}
	return l.o.Cache, l.errs
}

type job struct {
	p   provider.Provider
	src *provider.Source    // list its vaults
	c   *provider.Container // list its secrets
	key string              // where the list is cached
}

type result struct {
	job
	sources    []provider.Source
	containers []provider.Container
	objects    []provider.Object
	err        error
}

type lister struct {
	o       Options
	emit    func(Entry) error
	seen    map[string]bool // provider ID + link of entries emitted
	pending map[string]bool // cache keys queued or loading
	queue   []job
	active  int
	results chan result
	errs    []error
	stopped error // emit's error
}

func (l *lister) cacheable(p provider.Provider) bool { return p.Kind() != provider.KindRepo }

// load reports whether to load a list: a repo's always (from disk), else
// unless Cached, when it isn't cached or is older than MaxAge.
func (l *lister) load(p provider.Provider, cached bool, at time.Time) bool {
	if !l.cacheable(p) {
		return true
	}
	return !l.o.Cached && (!cached || l.o.Now().Sub(at) >= l.o.MaxAge)
}

func (l *lister) provider(p provider.Provider) {
	e, ok := l.o.Cache.Sources[p.ID()]
	if l.cacheable(p) && ok {
		for _, src := range sorted(e.Value, func(s provider.Source) string { return s.Name }) {
			l.source(p, src)
		}
	}
	if l.load(p, ok, e.At) {
		l.enqueue(job{p: p, key: p.ID()})
	}
}

// source emits src, and what's cached under it, and queues what's not.
func (l *lister) source(p provider.Provider, src provider.Source) {
	src.Provider, src.ProviderLabel = p.ID(), p.Label()
	if r, err := ref.ForSource(src); err == nil {
		l.put(p, Entry{Link: r.String(), Kind: "source", Provider: kindOf(src.Kind), Path: []string{src.Name}, Account: accountOf(src)})
	}
	row := cachefile.Row(p.ID(), src.ID)
	e, ok := l.o.Cache.Containers[row]
	if l.cacheable(p) && ok {
		l.containers(p, src, e.Value)
	}
	if l.load(p, ok, e.At) {
		l.enqueue(job{p: p, src: &src, key: row})
	}
}

func (l *lister) containers(p provider.Provider, src provider.Source, cs []provider.Container) {
	for _, c := range sorted(cs, func(c provider.Container) string { return c.Name }) {
		c.Source = src
		if r, err := ref.ForContainer(c, provider.TypeSecret); err == nil {
			l.put(p, Entry{Link: r.String(), Kind: "vault", Provider: kindOf(src.Kind), Path: []string{src.Name, c.Name}, Account: accountOf(src)})
		}
		key := cachefile.ObjectsKey(c, provider.TypeSecret)
		e, ok := l.o.Cache.Objects[key]
		if l.cacheable(p) {
			// Keys and certificates only as far as they were opened in
			// the TUI: Go to doesn't load them either.
			for _, t := range []provider.ObjectType{provider.TypeSecret, provider.TypeKey, provider.TypeCertificate} {
				l.objects(p, c, l.o.Cache.Objects[cachefile.ObjectsKey(c, t)].Value)
			}
		}
		if l.load(p, ok, e.At) {
			l.enqueue(job{p: p, c: &c, key: key})
		}
	}
}

func (l *lister) objects(p provider.Provider, c provider.Container, objs []provider.Object) {
	for _, o := range sorted(objs, func(o provider.Object) string { return o.Name }) {
		o.Container = c
		r, err := ref.ForObject(o, "")
		if err != nil {
			continue
		}
		l.put(p, Entry{Link: r.String(), Kind: strings.ToLower(string(o.Type)), Provider: kindOf(c.Source.Kind),
			Path: []string{c.Source.Name, c.Name, o.Name}, Account: accountOf(c.Source)})
	}
}

// put emits e, unless it was already (or emitting has failed).
func (l *lister) put(p provider.Provider, e Entry) {
	id := p.ID() + "\x00" + e.Link
	if l.stopped != nil || l.seen[id] {
		return
	}
	l.seen[id] = true
	l.stopped = l.emit(e)
}

func (l *lister) enqueue(j job) {
	if l.pending[j.key] {
		return
	}
	l.pending[j.key] = true
	l.queue = append(l.queue, j)
}

func (l *lister) start(ctx context.Context) {
	for l.stopped == nil && l.active < Workers && len(l.queue) > 0 {
		j := l.queue[0]
		l.queue = l.queue[1:]
		l.active++
		go func() { l.results <- l.fetch(ctx, j) }()
	}
}

func (l *lister) fetch(ctx context.Context, j job) result {
	kind := j.p.Kind()
	ctx, cancel := context.WithTimeout(ctx, l.o.Timeout(kind))
	defer cancel()
	r := result{job: j}
	switch {
	case j.c != nil:
		r.objects, r.err = j.p.Objects(ctx, *j.c, provider.TypeSecret)
	case j.src != nil:
		r.containers, r.err = j.p.Containers(ctx, *j.src)
	default:
		r.sources, r.err = j.p.Sources(ctx)
	}
	return r
}

func (l *lister) handle(r result) {
	delete(l.pending, r.key)
	now := l.o.Now()
	if r.err != nil {
		l.errs = append(l.errs, fmt.Errorf("%s: %w", r.label(), r.err))
		return
	}
	cache := l.cacheable(r.p)
	switch {
	case r.c != nil:
		if cache {
			l.o.Cache.Objects[r.key] = cachefile.Stamped[[]provider.Object]{At: now, Value: r.objects}
		}
		l.objects(r.p, *r.c, r.objects)
	case r.src != nil:
		r.containers = slices.Clone(r.containers) // the provider's may be shared
		for i := range r.containers {
			r.containers[i].Source = *r.src
		}
		if cache {
			l.o.Cache.Containers[r.key] = cachefile.Stamped[[]provider.Container]{At: now, Value: r.containers}
		}
		l.containers(r.p, *r.src, r.containers)
	default:
		if cache {
			l.o.Cache.Sources[r.key] = cachefile.Stamped[[]provider.Source]{At: now, Value: r.sources}
		}
		for _, src := range r.sources {
			l.source(r.p, src)
		}
	}
}

// label names a list that couldn't be loaded.
func (r result) label() string {
	switch {
	case r.c != nil:
		return r.c.Source.Name + " / " + r.c.Name
	case r.src != nil:
		return r.src.Name
	}
	return r.p.Label()
}

func kindOf(k provider.Kind) string { return strings.ToLower(string(k)) }

// accountOf is the account part of a source's provider label ("alice" in
// "Azure · alice"), or "" when there's a single account.
func accountOf(src provider.Source) string {
	account, ok := strings.CutPrefix(src.ProviderLabel, string(src.Kind)+" · ")
	if !ok {
		return ""
	}
	return account
}

func sorted[T any](s []T, name func(T) string) []T {
	s = slices.Clone(s)
	slices.SortStableFunc(s, func(a, b T) int {
		return cmp.Compare(strings.ToLower(name(a)), strings.ToLower(name(b)))
	})
	return s
}