chore: add mercury

mercury/pg/list-string.go

@@ -0,0 +1,120 @@
+package pg
+
+import (
+	"database/sql/driver"
+	"fmt"
+	"strings"
+	"unicode"
+	"unicode/utf8"
+)
+
+// ListStrings is a list of String values.
+type listString []string
+
+// Scan implements the Scanner interface for ListIDs.
+func (e *listString) Scan(value interface{}) (err error) {
+
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	case []rune:
+		str = string(v)
+	default:
+		return fmt.Errorf("array must be uint64, got: %T", value)
+	}
+
+	if e == nil {
+		*e = listString{}
+	}
+
+	str = trim(str, '{', '}')
+	if len(str) == 0 {
+		return nil
+	}
+
+	for _, s := range splitComma(string(str)) {
+		*e = append(*e, s)
+	}
+
+	return nil
+}
+
+// Value implements the driver Valuer interface for ListStrings.
+func (e listString) Value() (v driver.Value, err error) {
+	var b strings.Builder
+
+	if len(e) == 0 {
+		return "{}", nil
+	}
+
+	_, err = b.WriteRune('{')
+	if err != nil {
+		return
+	}
+
+	var arr []string
+	for _, s := range e {
+		arr = append(arr, `"`+s+`"`)
+	}
+	_, err = b.WriteString(strings.Join(arr, ","))
+	if err != nil {
+		return
+	}
+
+	_, err = b.WriteRune('}')
+	if err != nil {
+		return
+	}
+
+	return b.String(), nil
+}
+
+func splitComma(s string) []string {
+	lastQuote := rune(0)
+	f := func(c rune) bool {
+		switch {
+		case c == lastQuote:
+			lastQuote = rune(0)
+			return false
+		case lastQuote != rune(0):
+			return false
+		case unicode.In(c, unicode.Quotation_Mark):
+			lastQuote = c
+			return false
+		default:
+			return c == ','
+		}
+	}
+	lis := strings.FieldsFunc(s, f)
+
+	var out []string
+	for _, s := range lis {
+		s = trim(s, '"', '"')
+		out = append(out, s)
+	}
+
+	return out
+}
+
+func trim(s string, start, end rune) string {
+	r0, size0 := utf8.DecodeRuneInString(s)
+	if size0 == 0 {
+		return s
+	}
+	if r0 != start {
+		return s
+	}
+
+	r1, size1 := utf8.DecodeLastRuneInString(s)
+	if size1 == 0 {
+		return s
+	}
+	if r1 != end {
+		return s
+	}
+
+	return s[size0 : len(s)-size1]
+}

mercury/pg/notify.go

@@ -0,0 +1,45 @@
+package pg
+
+import (
+	"context"
+
+	"github.com/Masterminds/squirrel"
+
+	"go.sour.is/pkg/mercury"
+)
+
+// Notify stores the attributes for a registry space
+type Notify struct {
+	Name   string `json:"name" view:"mercury_notify_vw"`
+	Match  string `json:"match"`
+	Event  string `json:"event"`
+	Method string `json:"-" db:"method"`
+	URL    string `json:"-" db:"url"`
+}
+
+// GetNotify get list of rules
+func (pgm *postgresHandler) GetNotify(event string) (lis mercury.ListNotify, err error) {
+
+	rows, err := squirrel.Select("name", "match", "event", "method", "url").
+		From("mercury_notify_vw").
+		Where(squirrel.Eq{"event": event}).
+		PlaceholderFormat(squirrel.Dollar).
+		RunWith(pgm.db).
+		QueryContext(context.TODO())
+
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+	for rows.Next() {
+		var s mercury.Notify
+		err = rows.Scan(&s.Name, &s.Match, &s.Event, &s.Method, &s.URL)
+		if err != nil {
+			return nil, err
+		}
+		lis = append(lis, s)
+	}
+
+	return lis, rows.Err()
+}

mercury/pg/otel.go

@@ -0,0 +1,35 @@
+package pg
+
+import (
+	"database/sql"
+
+	"go.nhat.io/otelsql"
+	semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
+)
+
+func openDB(driver, dsn string) (*sql.DB, error) {
+	// Register the otelsql wrapper for the provided postgres driver.
+	driverName, err := otelsql.Register(driver,
+		otelsql.AllowRoot(),
+		otelsql.TraceQueryWithoutArgs(),
+		otelsql.TraceRowsClose(),
+		otelsql.TraceRowsAffected(),
+		// otelsql.WithDatabaseName("my_database"),        // Optional.
+		otelsql.WithSystem(semconv.DBSystemPostgreSQL), // Optional.
+	)
+	if err != nil {
+		return nil, err
+	}
+	
+	// Connect to a Postgres database using the postgres driver wrapper.
+	db, err := sql.Open(driverName, dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := otelsql.RecordStats(db); err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}

mercury/pg/postgres.go

@@ -0,0 +1,398 @@
+package pg
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log"
+	"strings"
+
+	sq "github.com/Masterminds/squirrel"
+	"go.sour.is/pkg/lg"
+	"go.sour.is/pkg/mercury"
+	"go.sour.is/pkg/rsql"
+	"golang.org/x/exp/maps"
+)
+
+type postgresHandler struct {
+	db *sql.DB
+}
+
+var (
+	_ mercury.GetIndex    = (*postgresHandler)(nil)
+	_ mercury.GetConfig   = (*postgresHandler)(nil)
+	_ mercury.GetRules    = (*postgresHandler)(nil)
+	_ mercury.WriteConfig = (*postgresHandler)(nil)
+)
+
+func Register() {
+	mercury.Registry.Register("sql", func(s *mercury.Space) any {
+		var dsn strings.Builder
+		var dbtype string
+		for _, c := range s.List {
+			if c.Name == "match" {
+				continue
+			}
+			if c.Name == "dbtype" {
+				dbtype = c.First()
+				continue
+			}
+			fmt.Fprintln(&dsn, c.Name, "=", c.First())
+		}
+
+		db, err := openDB(dbtype, dsn.String())
+		if err != nil {
+			return err
+		}
+		if err = db.Ping(); err != nil {
+			return err
+		}
+		return &postgresHandler{db}
+	})
+}
+
+type Space struct {
+	mercury.Space
+	ID uint64
+}
+type Value struct {
+	mercury.Value
+	ID uint64
+}
+
+func (p *postgresHandler) GetIndex(ctx context.Context, search mercury.NamespaceSearch, pgm *rsql.Program) (mercury.Config, error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	cols := rsql.GetDbColumns(mercury.Space{})
+	where, err := getWhere(search, cols)
+	if err != nil {
+		return nil, err
+	}
+	lis, err := p.listSpace(ctx, nil, where)
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+
+	config := make(mercury.Config, len(lis))
+	for i, s := range lis {
+		config[i] = &s.Space
+	}
+
+	return config, nil
+}
+
+func (p *postgresHandler) GetConfig(ctx context.Context, search mercury.NamespaceSearch, pgm *rsql.Program, fields []string) (mercury.Config, error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	idx, err := p.GetIndex(ctx, search, pgm)
+	if err != nil {
+		return nil, err
+	}
+	spaceMap := make(map[string]int, len(idx))
+	for u, s := range idx {
+		spaceMap[s.Space] = u
+	}
+
+	where, err := getWhere(search, rsql.GetDbColumns(mercury.Value{}))
+	if err != nil {
+		return nil, err
+	}
+	rows, err := sq.Select("space", "name", "seq", "notes", "tags", "values").
+		From("mercury_registry_vw").
+		Where(where).
+		OrderBy("space asc", "name asc").
+		PlaceholderFormat(sq.Dollar).
+		RunWith(p.db).
+		QueryContext(ctx)
+
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+
+	defer rows.Close()
+	for rows.Next() {
+		var (
+			s      mercury.Value
+			notes  listString
+			tags   listString
+			values listString
+		)
+		err = rows.Scan(&s.Space, &s.Name, &s.Seq, &notes, &tags, &values)
+		if err != nil {
+			return nil, err
+		}
+		s.Notes = notes
+		s.Tags = tags
+		s.Values = values
+		if u, ok := spaceMap[s.Space]; ok {
+			idx[u].List = append(idx[u].List, s)
+		}
+	}
+
+	err = rows.Err()
+	span.RecordError(err)
+
+	span.AddEvent(fmt.Sprint("read index ", len(idx)))
+	return idx, err
+}
+
+func (p *postgresHandler) listSpace(ctx context.Context, tx sq.BaseRunner, where sq.Sqlizer) ([]*Space, error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	if tx == nil {
+		tx = p.db
+	}
+
+	rows, err := sq.Select("id", "space", "tags", "notes").
+		From("mercury_spaces").
+		Where(where).
+		OrderBy("space asc").
+		PlaceholderFormat(sq.Dollar).
+		RunWith(tx).
+		QueryContext(ctx)
+
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	defer rows.Close()
+
+	var lis []*Space
+	for rows.Next() {
+		var (
+			s     Space
+			name  string
+			tags  = &listString{}
+			notes = &listString{}
+		)
+		err = rows.Scan(&s.ID, &name, tags, notes)
+		if err != nil {
+			return nil, err
+		}
+		s.Space.Space = name
+		s.Space.Notes = *notes
+		s.Space.Tags = *tags
+		lis = append(lis, &s)
+	}
+
+	err = rows.Err()
+	span.RecordError(err)
+
+	span.AddEvent(fmt.Sprint("read config ", len(lis)))
+	return lis, err
+}
+
+// WriteConfig writes a config map to database
+func (p *postgresHandler) WriteConfig(ctx context.Context, config mercury.Config) (err error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	// Delete spaces that are present in input but are empty.
+	deleteSpaces := make(map[string]struct{})
+
+	// get names of each space
+	var names = make(map[string]int)
+	for i, v := range config {
+		names[v.Space] = i
+
+		if len(v.Tags) == 0 && len(v.Notes) == 0 && len(v.List) == 0 {
+			deleteSpaces[v.Space] = struct{}{}
+		}
+	}
+
+	tx, err := p.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err != nil && tx != nil {
+			tx.Rollback()
+		}
+	}()
+
+	// get current spaces
+	lis, err := p.listSpace(ctx, tx, sq.Eq{"space": maps.Keys(names)})
+	if err != nil {
+		return
+	}
+
+	// determine which are being updated
+	var deleteIDs []uint64
+	var updateIDs []uint64
+	var currentNames = make(map[string]struct{}, len(lis))
+	var updateSpaces []*mercury.Space
+	var insertSpaces []*mercury.Space
+
+	for _, s := range lis {
+		spaceName := s.Space.Space
+		currentNames[spaceName] = struct{}{}
+
+		if _, ok := deleteSpaces[spaceName]; ok {
+			deleteIDs = append(deleteIDs, s.ID)
+			continue
+		}
+
+		updateSpaces = append(updateSpaces, config[names[spaceName]])
+		updateIDs = append(updateIDs, s.ID)
+	}
+	for _, s := range config {
+		spaceName := s.Space
+		if _, ok := currentNames[spaceName]; !ok {
+			insertSpaces = append(insertSpaces, s)
+		}
+	}
+
+	// delete spaces
+	if ids := deleteIDs; len(ids) > 0 {
+		_, err = sq.Delete("mercury_spaces").Where(sq.Eq{"id": ids}).RunWith(tx).PlaceholderFormat(sq.Dollar).ExecContext(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
+	// delete values
+	if ids := append(updateIDs, deleteIDs...); len(ids) > 0 {
+		_, err = sq.Delete("mercury_values").Where(sq.Eq{"id": ids}).RunWith(tx).PlaceholderFormat(sq.Dollar).ExecContext(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
+	var newValues []*Value
+
+	// update spaces
+	for i, u := range updateSpaces {
+		_, err := sq.Update("mercury_spaces").
+			Where(sq.Eq{"id": updateIDs[i]}).
+			Set("tags", listString(u.Tags)).
+			Set("notes", listString(u.Notes)).
+			PlaceholderFormat(sq.Dollar).
+			RunWith(tx).ExecContext(ctx)
+
+		if err != nil {
+			return err
+		}
+		// log.Debugf("UPDATED %d SPACES", len(updateSpaces))
+		for _, v := range u.List {
+			newValues = append(newValues, &Value{Value: v, ID: updateIDs[i]})
+		}
+	}
+
+	// insert spaces
+	for _, s := range insertSpaces {
+		var id uint64
+		query := sq.Insert("mercury_spaces").
+			PlaceholderFormat(sq.Dollar).
+			Columns("space", "tags", "notes").
+			Values(s.Space, listString(s.Tags), listString(s.Notes)).
+			Suffix("RETURNING \"id\"")
+		err := query.
+			RunWith(tx).
+			QueryRowContext(ctx).
+			Scan(&id)
+		if err != nil {
+			s, v, _ := query.ToSql()
+			log.Println(s, v, err)
+			return err
+		}
+		for _, v := range s.List {
+			newValues = append(newValues, &Value{Value: v, ID: id})
+		}
+	}
+
+	// write all values to db.
+	err = p.writeValues(ctx, tx, newValues)
+	// log.Debugf("WROTE %d ATTRS", len(attrs))
+
+	tx.Commit()
+	tx = nil
+
+	return
+}
+
+// writeValues writes the values to db
+func (p *postgresHandler) writeValues(ctx context.Context, tx sq.BaseRunner, lis []*Value) (err error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	if len(lis) == 0 {
+		return nil
+	}
+
+	newInsert := func() sq.InsertBuilder {
+		return sq.Insert("mercury_values").
+			RunWith(tx).
+			PlaceholderFormat(sq.Dollar).
+			Columns(
+				"id",
+				"seq",
+				"name",
+				"values",
+				"notes",
+				"tags",
+			)
+	}
+	chunk := int(65000 / 3)
+	insert := newInsert()
+	for i, s := range lis {
+		insert = insert.Values(
+			s.ID,
+			s.Seq,
+			s.Name,
+			listString(s.Values),
+			listString(s.Notes),
+			listString(s.Tags),
+		)
+		// log.Debug(s.Name)
+
+		if i > 0 && i%chunk == 0 {
+			// log.Debugf("inserting %v rows into %v", i%chunk, d.Table)
+			// log.Debug(insert.ToSql())
+
+			_, err = insert.ExecContext(ctx)
+			if err != nil {
+				// log.Error(err)
+				return
+			}
+
+			insert = newInsert()
+		}
+	}
+	if len(lis)%chunk > 0 {
+		// log.Debugf("inserting %v rows into %v", len(lis)%chunk, d.Table)
+		// log.Debug(insert.ToSql())
+
+		_, err = insert.ExecContext(ctx)
+		if err != nil {
+			// log.Error(err)
+			return
+		}
+	}
+
+	return
+}
+
+func getWhere(search mercury.NamespaceSearch, d *rsql.DbColumns) (sq.Sqlizer, error) {
+	var where sq.Or
+	space, err := d.Col("space")
+	if err != nil {
+		return nil, err
+	}
+	for _, m := range search {
+		switch m.(type) {
+		case mercury.NamespaceNode:
+			where = append(where, sq.Eq{space: m.Value()})
+		case mercury.NamespaceStar:
+			where = append(where, sq.Like{space: m.Value()})
+		case mercury.NamespaceTrace:
+			e := sq.Expr(`? LIKE concat(`+space+`, '%')`, m.Value())
+			where = append(where, e)
+		}
+	}
+	return where, nil
+}

mercury/pg/rules.go

@@ -0,0 +1,92 @@
+package pg
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Masterminds/squirrel"
+	"go.sour.is/pkg/lg"
+	"go.sour.is/pkg/mercury"
+)
+
+type grouper interface {
+	GetGroups() []string
+}
+
+// GetRules get list of rules
+func (p *postgresHandler) GetRules(ctx context.Context, user mercury.Ident) (lis mercury.Rules, err error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	var ids []string
+	ids = append(ids, "U-"+user.GetIdentity())
+	switch u := user.(type) {
+	case grouper:
+		for _, g := range u.GetGroups() {
+			ids = append(ids, "G-"+g)
+		}
+	}
+	if groups, err := p.getGroups(ctx, user.GetIdentity()); err != nil {
+		for _, g := range groups {
+			ids = append(ids, "G-"+g)
+		}
+	}
+
+	query := squirrel.Select("role", "type", "match").
+		From("mercury_rules_vw").
+		Where(squirrel.Eq{"id": ids}).
+		PlaceholderFormat(squirrel.Dollar)
+	rows, err := query.
+		RunWith(p.db).
+		QueryContext(ctx)
+
+	if err != nil {
+		span.RecordError(err)
+		return nil, err
+	}
+
+	defer rows.Close()
+	for rows.Next() {
+		var s mercury.Rule
+		err = rows.Scan(&s.Role, &s.Type, &s.Match)
+		if err != nil {
+			span.RecordError(err)
+			return nil, err
+		}
+		lis = append(lis, s)
+	}
+	err = rows.Err()
+	span.RecordError(err)
+
+	span.AddEvent(fmt.Sprint("read rules ", len(lis)))
+	return lis, err
+}
+
+// getGroups get list of groups
+func (pgm *postgresHandler) getGroups(ctx context.Context, user string) (lis []string, err error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	rows, err := squirrel.Select("group_id").
+		From("mercury_groups_vw").
+		Where(squirrel.Eq{"user_id": user}).
+		PlaceholderFormat(squirrel.Dollar).
+		RunWith(pgm.db).
+		QueryContext(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+	for rows.Next() {
+		var s string
+		err = rows.Scan(&s)
+		if err != nil {
+			return nil, err
+		}
+		lis = append(lis, s)
+	}
+
+	return lis, rows.Err()
+}