131 lines
6.5 KiB
Go
131 lines
6.5 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"git.cin.su/m/Assistant/contracts/capability"
|
|
"git.cin.su/m/Assistant/contracts/identity"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var ErrCapabilityHealthNotFound = errors.New("capability health not found")
|
|
var ErrCapabilityBindingNotFound = errors.New("capability binding not found")
|
|
var ErrCapabilityBindingConflict = errors.New("capability binding conflict")
|
|
|
|
type CapabilityRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func OpenCapability(ctx context.Context, config ContinuityConfig) (*CapabilityRepository, 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 &CapabilityRepository{pool: pool}, nil
|
|
}
|
|
|
|
func (r *CapabilityRepository) Close() { r.pool.Close() }
|
|
|
|
func (r *CapabilityRepository) SaveHealth(ctx context.Context, health capability.Health) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO capability.health (capability_id, state, reason, revision, semantic_version, observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (capability_id) DO UPDATE SET state=EXCLUDED.state, reason=EXCLUDED.reason, revision=EXCLUDED.revision, semantic_version=EXCLUDED.semantic_version, observed_at_unix_millis=EXCLUDED.observed_at_unix_millis WHERE capability.health.revision < EXCLUDED.revision`, health.CapabilityID, health.State, health.Reason, health.Revision, health.SemanticVersion, health.ObservedAtUnixMillis)
|
|
return err
|
|
}
|
|
|
|
func (r *CapabilityRepository) LoadHealth(ctx context.Context, id capability.CapabilityID) (capability.Health, error) {
|
|
var health capability.Health
|
|
err := r.pool.QueryRow(ctx, `SELECT capability_id, state, reason, revision, semantic_version, observed_at_unix_millis FROM capability.health WHERE capability_id=$1`, id).Scan(&health.CapabilityID, &health.State, &health.Reason, &health.Revision, &health.SemanticVersion, &health.ObservedAtUnixMillis)
|
|
if err != nil {
|
|
return capability.Health{}, fmt.Errorf("load capability health: %w", err)
|
|
}
|
|
return health, nil
|
|
}
|
|
|
|
func (r *CapabilityRepository) RebuildHealthProjection(ctx context.Context) (int64, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
if _, err = tx.Exec(ctx, `DELETE FROM capability.health_projection`); err != nil {
|
|
return 0, err
|
|
}
|
|
result, err := tx.Exec(ctx, `INSERT INTO capability.health_projection (capability_id, state, reason, revision, semantic_version, observed_at_unix_millis) SELECT capability_id, state, reason, revision, semantic_version, observed_at_unix_millis FROM capability.health`)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
func (r *CapabilityRepository) SaveBinding(ctx context.Context, binding capability.Binding) error {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
result, err := tx.Exec(ctx, `INSERT INTO capability.notification_bindings (
|
|
binding_id,capability_id,owner_principal_id,revision,state,https_origin,delivery_path,receipt_path,
|
|
protocol,secret_ref,verified_at_unix_millis,health)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT DO NOTHING`,
|
|
binding.ID, binding.CapabilityID, binding.OwnerPrincipalID, binding.Revision, binding.State,
|
|
binding.HTTPSOrigin, binding.DeliveryPath, binding.ReceiptPath, binding.Protocol, binding.SecretRef,
|
|
binding.VerifiedAtUnixMillis, binding.Health)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if result.RowsAffected() == 0 {
|
|
existing, loadErr := r.LoadBinding(ctx, binding.ID, binding.Revision)
|
|
if loadErr != nil || existing != binding {
|
|
return ErrCapabilityBindingConflict
|
|
}
|
|
}
|
|
if binding.State == capability.BindingActive {
|
|
result, err = tx.Exec(ctx, `INSERT INTO capability.notification_binding_heads (capability_id,owner_principal_id,binding_id,revision)
|
|
VALUES ($1,$2,$3,$4) ON CONFLICT (capability_id,owner_principal_id) DO UPDATE SET binding_id=EXCLUDED.binding_id,revision=EXCLUDED.revision
|
|
WHERE capability.notification_binding_heads.revision < EXCLUDED.revision`, binding.CapabilityID, binding.OwnerPrincipalID, binding.ID, binding.Revision)
|
|
if err != nil || result.RowsAffected() != 1 {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return ErrCapabilityBindingConflict
|
|
}
|
|
} else {
|
|
_, err = tx.Exec(ctx, `DELETE FROM capability.notification_binding_heads WHERE capability_id=$1 AND owner_principal_id=$2 AND binding_id=$3 AND revision < $4`, binding.CapabilityID, binding.OwnerPrincipalID, binding.ID, binding.Revision)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (r *CapabilityRepository) LoadBinding(ctx context.Context, id capability.CapabilityBindingID, revision int64) (capability.Binding, error) {
|
|
var binding capability.Binding
|
|
err := r.pool.QueryRow(ctx, `SELECT binding_id,capability_id,owner_principal_id,revision,state,https_origin,delivery_path,receipt_path,
|
|
protocol,secret_ref,verified_at_unix_millis,health FROM capability.notification_bindings WHERE binding_id=$1 AND revision=$2`, id, revision).
|
|
Scan(&binding.ID, &binding.CapabilityID, &binding.OwnerPrincipalID, &binding.Revision, &binding.State,
|
|
&binding.HTTPSOrigin, &binding.DeliveryPath, &binding.ReceiptPath, &binding.Protocol, &binding.SecretRef,
|
|
&binding.VerifiedAtUnixMillis, &binding.Health)
|
|
if err != nil {
|
|
return capability.Binding{}, fmt.Errorf("load capability binding: %w", err)
|
|
}
|
|
return binding, nil
|
|
}
|
|
|
|
func (r *CapabilityRepository) LoadActiveBinding(ctx context.Context, capabilityID capability.CapabilityID, owner identity.PrincipalID) (capability.Binding, error) {
|
|
var binding capability.Binding
|
|
err := r.pool.QueryRow(ctx, `SELECT b.binding_id,b.capability_id,b.owner_principal_id,b.revision,b.state,b.https_origin,b.delivery_path,b.receipt_path,
|
|
b.protocol,b.secret_ref,b.verified_at_unix_millis,b.health FROM capability.notification_binding_heads h
|
|
JOIN capability.notification_bindings b ON b.binding_id=h.binding_id AND b.revision=h.revision
|
|
WHERE h.capability_id=$1 AND h.owner_principal_id=$2 AND b.state=$3`, capabilityID, owner, capability.BindingActive).
|
|
Scan(&binding.ID, &binding.CapabilityID, &binding.OwnerPrincipalID, &binding.Revision, &binding.State,
|
|
&binding.HTTPSOrigin, &binding.DeliveryPath, &binding.ReceiptPath, &binding.Protocol, &binding.SecretRef,
|
|
&binding.VerifiedAtUnixMillis, &binding.Health)
|
|
if err != nil {
|
|
return capability.Binding{}, fmt.Errorf("load active capability binding: %w", err)
|
|
}
|
|
return binding, nil
|
|
}
|