59 lines
3.0 KiB
Go
59 lines
3.0 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"git.cin.su/m/Assistant/contracts/continuity"
|
|
"git.cin.su/m/Assistant/contracts/identity"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var ErrProspectiveNotFound = errors.New("prospective item not found")
|
|
|
|
type ContinuityConfig struct{ Host, Port, Database, User, Password string }
|
|
type ContinuityRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func OpenContinuity(ctx context.Context, config ContinuityConfig) (*ContinuityRepository, error) {
|
|
dsn := fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s", config.Host, config.Port, config.Database, config.User, config.Password)
|
|
pool, err := pgxpool.New(ctx, dsn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &ContinuityRepository{pool: pool}, nil
|
|
}
|
|
func (r *ContinuityRepository) Close() { r.pool.Close() }
|
|
func (r *ContinuityRepository) Save(ctx context.Context, item continuity.ProspectiveItem) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO continuity.prospective_items (prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, trigger_at_unix_millis, state, revision) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT (prospective_item_id) DO UPDATE SET state=EXCLUDED.state, revision=EXCLUDED.revision`, item.ID, item.CommitmentID, item.OwnerPrincipalID, item.ScopeID, item.GoalID, item.TriggerAtUnixMillis, item.State, item.Revision)
|
|
return err
|
|
}
|
|
func (r *ContinuityRepository) Load(ctx context.Context, id continuity.ProspectiveItemID) (continuity.ProspectiveItem, error) {
|
|
var item continuity.ProspectiveItem
|
|
err := r.pool.QueryRow(ctx, `SELECT prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, trigger_at_unix_millis, state, revision FROM continuity.prospective_items WHERE prospective_item_id=$1`, id).Scan(&item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, &item.TriggerAtUnixMillis, &item.State, &item.Revision)
|
|
if err != nil {
|
|
return continuity.ProspectiveItem{}, fmt.Errorf("load prospective item: %w", err)
|
|
}
|
|
return item, nil
|
|
}
|
|
|
|
func (r *ContinuityRepository) ByOwner(ctx context.Context, owner identity.PrincipalID, limit int) ([]continuity.ProspectiveItem, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, trigger_at_unix_millis, state, revision FROM continuity.prospective_items WHERE owner_principal_id=$1 ORDER BY trigger_at_unix_millis, prospective_item_id LIMIT $2`, owner, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list prospective items by owner: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
items := make([]continuity.ProspectiveItem, 0)
|
|
for rows.Next() {
|
|
var item continuity.ProspectiveItem
|
|
if err := rows.Scan(&item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, &item.TriggerAtUnixMillis, &item.State, &item.Revision); err != nil {
|
|
return nil, fmt.Errorf("scan prospective item: %w", err)
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("list prospective items by owner: %w", err)
|
|
}
|
|
return items, nil
|
|
}
|