upstream-ory-keto/internal/driver/config/opl_config_namespace_watcher.go

// Copyright © 2023 Ory Corp
// SPDX-License-Identifier: Apache-2.0

package config

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"sync"
	"time"

	"github.com/dgraph-io/ristretto/v2"
	"github.com/ory/x/fetcher"
	"github.com/ory/x/logrusx"
	"github.com/ory/x/urlx"
	"github.com/ory/x/watcherx"
	"github.com/pkg/errors"
	"golang.org/x/sync/singleflight"

	"github.com/ory/keto/internal/namespace"
	"github.com/ory/keto/schema"
)

type (
	configFiles struct {
		byPath map[string]io.Reader
		sync.Mutex
	}

	oplConfigWatcher struct {
		logger *logrusx.Logger
		target string
		files  configFiles

		memoryNamespaceManager
	}
)

var (
	_ namespace.Manager = (*oplConfigWatcher)(nil)

	// cache holds the parsed namespaces per remote OPL location. Content at a location is
	// assumed to be immutable, therefore, a change to the OPL should create a new file location,
	// therefore a new cache entry.
	cache, _ = ristretto.NewCache(&ristretto.Config[string, []*namespace.Namespace]{
		MaxCost:     20_000_000, // 20 MB of estimated heap, each item ca. 10 KB => max 2000 items
		NumCounters: 20_000,     // max 2000 items => 20000 counters
		BufferItems: 64,
		Cost:        namespace.EstimatedSize,
	})

	fetchGroup singleflight.Group
)

func newOPLConfigWatcher(ctx context.Context, l *logrusx.Logger, newFetcher func() *fetcher.Fetcher, target string) (*oplConfigWatcher, error) {
	nw := &oplConfigWatcher{
		logger:                 l,
		target:                 target,
		files:                  configFiles{byPath: make(map[string]io.Reader)},
		memoryNamespaceManager: *NewMemoryNamespaceManager(),
	}

	targetUrl, err := urlx.Parse(target)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	switch targetUrl.Scheme {
	case "file", "":
		return nw, watchTarget(ctx, target, nw, l)
	case "base64":
		file, err := newFetcher().FetchContext(ctx, target)
		if err != nil {
			return nil, err
		}
		nw.files.Lock()
		defer nw.files.Unlock()
		nw.files.byPath[targetUrl.String()] = file
		nw.parseFiles()
		return nw, err
	case "http", "https":
		if namespaces, ok := cache.Get(target); ok {
			nw.set(namespaces)
			return nw, nil
		}
		v, err, _ := fetchGroup.Do(target, func() (any, error) {
			// Detach from the leader's context so that its cancellation does not
			// fail the concurrent callers sharing this fetch.
			ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second)
			defer cancel()

			buf, err := newFetcher().FetchContext(ctx, target)
			if err != nil {
				return nil, err
			}
			b := buf.Bytes()
			nw.files.Lock()
			defer nw.files.Unlock()
			nw.files.byPath[targetUrl.String()] = bytes.NewReader(b)
			namespaces, ok := nw.parseFiles()
			if !ok {
				// Cache a parse failure as an empty result, so that a broken
				// OPL file does not trigger a fetch per construction.
				namespaces = []*namespace.Namespace{}
			}
			// Cost 0 lets the cache's Cost function compute the entry size.
			cache.SetWithTTL(target, namespaces, 0, 30*time.Minute)
			return namespaces, nil
		})
		if err != nil {
			return nil, err
		}
		if namespaces, ok := v.([]*namespace.Namespace); ok {
			nw.set(namespaces)
		}
		return nw, nil
	default:
		return nil, fmt.Errorf("unexpected url scheme: %q", targetUrl.Scheme)
	}
}

func (nw *oplConfigWatcher) handleChange(e *watcherx.ChangeEvent) {
	// the lock is acquired before parsing to ensure that the getters are
	// waiting for the updated values
	nw.files.Lock()
	defer nw.files.Unlock()
	nw.files.byPath[e.Source()] = e.Reader()
	nw.parseFiles()
}

func (nw *oplConfigWatcher) handleRemove(e *watcherx.RemoveEvent) {
	nw.files.Lock()
	defer nw.files.Unlock()
	delete(nw.files.byPath, e.Source())
	nw.parseFiles()
}

func (nw *oplConfigWatcher) handleError(e *watcherx.ErrorEvent) {
	nw.logger.
		WithError(e).
		Errorf("Received error while watching OPL config files at target %s.",
			nw.target)
}

// parseFiles loops through all files, parsing each and getting the namespaces.
// It then sets and returns the namespaces only if there were no errors.
//
// The caller must  hold the lock to nw.files.
func (nw *oplConfigWatcher) parseFiles() ([]*namespace.Namespace, bool) {
	var (
		namespaces = make([]*namespace.Namespace, 0)
		errs       []error
	)
	for _, reader := range nw.files.byPath {
		content, err := io.ReadAll(reader)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		nn, ee := schema.Parse(string(content))
		for _, e := range ee {
			errs = append(errs, e)
		}
		for _, n := range nn {
			namespaces = append(namespaces, &n)
		}
	}
	if len(errs) > 0 {
		for _, err := range errs {
			nw.logger.
				WithError(err).
				Errorf("Failed to parse OPL config files at target %s.",
					nw.target)
		}
		return nil, false
	}
	nw.set(namespaces)
	return namespaces, true
}