package sqlite import ( "context" "database/sql" "encoding/json" "fmt" "time" "forge.cadoles.com/Cadoles/emissary/internal/datastore" "github.com/lestrrat-go/jwx/v2/jwk" "github.com/pkg/errors" "gitlab.com/wpetit/goweb/logger" ) type AgentRepository struct { db *sql.DB } // DeleteSpec implements datastore.AgentRepository. func (r *AgentRepository) DeleteSpec(ctx context.Context, agentID datastore.AgentID, name string) error { query := `DELETE FROM specs WHERE agent_id = $1 AND name = $2` _, err := r.db.ExecContext(ctx, query, agentID, name) if err != nil { return errors.WithStack(err) } return nil } // GetSpecs implements datastore.AgentRepository. func (r *AgentRepository) GetSpecs(ctx context.Context, agentID datastore.AgentID) ([]*datastore.Spec, error) { specs := make([]*datastore.Spec, 0) query := ` SELECT id, name, revision, data, created_at, updated_at FROM specs WHERE agent_id = $1 ` rows, err := r.db.QueryContext(ctx, query, agentID) if err != nil { return nil, errors.WithStack(err) } defer rows.Close() for rows.Next() { spec := &datastore.Spec{} data := JSONMap{} if err := rows.Scan(&spec.ID, &spec.Name, &spec.Revision, &data, &spec.CreatedAt, &spec.UpdatedAt); err != nil { return nil, errors.WithStack(err) } spec.Data = data specs = append(specs, spec) } return specs, nil } // UpdateSpec implements datastore.AgentRepository. func (r *AgentRepository) UpdateSpec(ctx context.Context, agentID datastore.AgentID, name string, revision int, data map[string]any) (*datastore.Spec, error) { spec := &datastore.Spec{} err := r.withTx(ctx, func(tx *sql.Tx) error { now := time.Now().UTC() query := ` INSERT INTO specs (agent_id, name, revision, data, created_at, updated_at) VALUES($1, $2, $3, $4, $5, $5) ON CONFLICT (agent_id, name) DO UPDATE SET data = $4, updated_at = $5, revision = specs.revision + 1 WHERE revision = $3 RETURNING "id", "name", "revision", "data", "created_at", "updated_at" ` args := []any{agentID, name, revision, JSONMap(data), now} logger.Debug(ctx, "executing query", logger.F("query", query), logger.F("args", args)) row := tx.QueryRowContext(ctx, query, args...) data := JSONMap{} err := row.Scan(&spec.ID, &spec.Name, &spec.Revision, &data, &spec.CreatedAt, &spec.UpdatedAt) if err != nil { if errors.Is(err, sql.ErrNoRows) { return errors.WithStack(datastore.ErrUnexpectedRevision) } return errors.WithStack(err) } spec.Data = data return nil }) if err != nil { return nil, errors.WithStack(err) } return spec, nil } // Query implements datastore.AgentRepository. func (r *AgentRepository) Query(ctx context.Context, opts ...datastore.AgentQueryOptionFunc) ([]*datastore.Agent, int, error) { options := &datastore.AgentQueryOptions{} for _, fn := range opts { fn(options) } agents := make([]*datastore.Agent, 0) count := 0 err := r.withTx(ctx, func(tx *sql.Tx) error { query := `SELECT id, thumbprint, status, created_at, updated_at FROM agents` limit := 10 if options.Limit != nil { limit = *options.Limit } offset := 0 if options.Offset != nil { offset = *options.Offset } filters := "" paramIndex := 3 args := []any{offset, limit} if options.IDs != nil && len(options.IDs) > 0 { filter, newArgs, newParamIndex := inFilter("id", paramIndex, options.IDs) filters += filter paramIndex = newParamIndex args = append(args, newArgs...) } if options.Thumbprints != nil && len(options.Thumbprints) > 0 { if filters != "" { filters += " AND " } filter, newArgs, newParamIndex := inFilter("thumbprint", paramIndex, options.Thumbprints) filters += filter paramIndex = newParamIndex args = append(args, newArgs...) } if options.Statuses != nil && len(options.Statuses) > 0 { if filters != "" { filters += " AND " } filter, newArgs, _ := inFilter("status", paramIndex, options.Statuses) filters += filter args = append(args, newArgs...) } if filters != "" { filters = ` WHERE ` + filters } query += filters + ` LIMIT $2 OFFSET $1` logger.Debug(ctx, "executing query", logger.F("query", query), logger.F("args", args)) rows, err := r.db.QueryContext(ctx, query, args...) if err != nil { return errors.WithStack(err) } defer rows.Close() for rows.Next() { agent := &datastore.Agent{} metadata := JSONMap{} if err := rows.Scan(&agent.ID, &agent.Thumbprint, &agent.Status, &agent.CreatedAt, &agent.UpdatedAt); err != nil { return errors.WithStack(err) } agent.Metadata = metadata agents = append(agents, agent) } row := tx.QueryRowContext(ctx, `SELECT count(id) FROM agents `+filters, args...) if err := row.Scan(&count); err != nil { return errors.WithStack(err) } return nil }) if err != nil { return nil, 0, errors.WithStack(err) } return agents, count, nil } // Create implements datastore.AgentRepository func (r *AgentRepository) Create(ctx context.Context, thumbprint string, keySet jwk.Set, metadata map[string]any) (*datastore.Agent, error) { agent := &datastore.Agent{} err := r.withTx(ctx, func(tx *sql.Tx) error { query := `SELECT count(id) FROM agents WHERE thumbprint = $1` row := tx.QueryRowContext(ctx, query, thumbprint) var count int if err := row.Scan(&count); err != nil { return errors.WithStack(err) } if count > 0 { return errors.WithStack(datastore.ErrAlreadyExist) } now := time.Now().UTC() query = ` INSERT INTO agents (thumbprint, keyset, metadata, status, created_at, updated_at) VALUES($1, $2, $3, $4, $5, $5) RETURNING "id", "thumbprint", "keyset", "metadata", "status", "created_at", "updated_at" ` rawKeySet, err := json.Marshal(keySet) if err != nil { return errors.WithStack(err) } row = tx.QueryRowContext( ctx, query, thumbprint, rawKeySet, JSONMap(metadata), datastore.AgentStatusPending, now, ) metadata := JSONMap{} err = row.Scan(&agent.ID, &agent.Thumbprint, &rawKeySet, &metadata, &agent.Status, &agent.CreatedAt, &agent.UpdatedAt) if err != nil { return errors.WithStack(err) } agent.Metadata = metadata keySet, err = jwk.Parse(rawKeySet) if err != nil { return errors.WithStack(err) } agent.KeySet = &datastore.SerializableKeySet{keySet} return nil }) if err != nil { return nil, errors.WithStack(err) } return agent, nil } // Delete implements datastore.AgentRepository func (r *AgentRepository) Delete(ctx context.Context, id datastore.AgentID) error { err := r.withTx(ctx, func(tx *sql.Tx) error { query := `DELETE FROM agents WHERE id = $1` _, err := r.db.ExecContext(ctx, query, id) if err != nil { return errors.WithStack(err) } query = `DELETE FROM specs WHERE agent_id = $1` _, err = r.db.ExecContext(ctx, query, id) if err != nil { return errors.WithStack(err) } return nil }) if err != nil { return errors.WithStack(err) } return nil } // Get implements datastore.AgentRepository func (r *AgentRepository) Get(ctx context.Context, id datastore.AgentID) (*datastore.Agent, error) { agent := &datastore.Agent{ ID: id, } err := r.withTx(ctx, func(tx *sql.Tx) error { query := ` SELECT "id", "thumbprint", "keyset", "metadata", "status", "created_at", "updated_at" FROM agents WHERE id = $1 ` row := r.db.QueryRowContext(ctx, query, id) metadata := JSONMap{} var rawKeySet []byte if err := row.Scan(&agent.ID, &agent.Thumbprint, &rawKeySet, &metadata, &agent.Status, &agent.CreatedAt, &agent.UpdatedAt); err != nil { if errors.Is(err, sql.ErrNoRows) { return datastore.ErrNotFound } return errors.WithStack(err) } agent.Metadata = metadata keySet := jwk.NewSet() if err := json.Unmarshal(rawKeySet, &keySet); err != nil { return errors.WithStack(err) } agent.KeySet = &datastore.SerializableKeySet{keySet} return nil }) if err != nil { return nil, errors.WithStack(err) } return agent, nil } // Update implements datastore.AgentRepository func (r *AgentRepository) Update(ctx context.Context, id datastore.AgentID, opts ...datastore.AgentUpdateOptionFunc) (*datastore.Agent, error) { options := &datastore.AgentUpdateOptions{} for _, fn := range opts { fn(options) } agent := &datastore.Agent{} err := r.withTx(ctx, func(tx *sql.Tx) error { query := ` UPDATE agents SET updated_at = $2 ` now := time.Now().UTC() args := []any{ id, now, } index := 3 if options.Status != nil { query += fmt.Sprintf(`, status = $%d`, index) args = append(args, *options.Status) index++ } if options.KeySet != nil { rawKeySet, err := json.Marshal(*options.KeySet) if err != nil { return errors.WithStack(err) } query += fmt.Sprintf(`, keyset = $%d`, index) args = append(args, rawKeySet) index++ } if options.Thumbprint != nil { query += fmt.Sprintf(`, thumbprint = $%d`, index) args = append(args, *options.Thumbprint) index++ } if options.Metadata != nil { query += fmt.Sprintf(`, metadata = $%d`, index) args = append(args, JSONMap(*options.Metadata)) index++ } query += ` WHERE id = $1 RETURNING "id", "thumbprint", "keyset", "metadata", "status", "created_at", "updated_at" ` row := tx.QueryRowContext(ctx, query, args...) metadata := JSONMap{} var rawKeySet []byte if err := row.Scan(&agent.ID, &agent.Thumbprint, &rawKeySet, &metadata, &agent.Status, &agent.CreatedAt, &agent.UpdatedAt); err != nil { if errors.Is(err, sql.ErrNoRows) { return datastore.ErrNotFound } return errors.WithStack(err) } agent.Metadata = metadata keySet := jwk.NewSet() if err := json.Unmarshal(rawKeySet, &keySet); err != nil { return errors.WithStack(err) } agent.KeySet = &datastore.SerializableKeySet{keySet} return nil }) if err != nil { return nil, errors.WithStack(err) } return agent, nil } func (r *AgentRepository) withTx(ctx context.Context, fn func(*sql.Tx) error) error { tx, err := r.db.Begin() if err != nil { return errors.WithStack(err) } defer func() { if err := tx.Rollback(); err != nil { if errors.Is(err, sql.ErrTxDone) { return } logger.Error(ctx, "could not rollback transaction", logger.E(errors.WithStack(err))) } }() if err := fn(tx); err != nil { return errors.WithStack(err) } if err := tx.Commit(); err != nil { return errors.WithStack(err) } return nil } func NewAgentRepository(db *sql.DB) *AgentRepository { return &AgentRepository{db} } var _ datastore.AgentRepository = &AgentRepository{} func inFilter[T any](column string, paramIndex int, items []T) (string, []any, int) { args := make([]any, 0, len(items)) filter := fmt.Sprintf("%s in (", column) for idx, item := range items { if idx != 0 { filter += "," } filter += fmt.Sprintf("$%d", paramIndex) paramIndex++ args = append(args, item) } filter += ")" return filter, args, paramIndex }