Nereus/apps/api/store.go

package main

import (
	"context"
	_ "embed"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"go.opentelemetry.io/otel"
)

//go:embed migrations/001_init.sql
var migrationSQL string

type store interface {
	Ping(context.Context) error
	Migrate(context.Context) error
	ListBuoys(context.Context) ([]buoy, error)
	CreateBuoy(context.Context, buoy) (buoy, error)
	GetBuoy(context.Context, string) (buoy, error)
	DeleteBuoy(context.Context, string) error
	ListReadings(context.Context, string, *time.Time, *time.Time, int, int) ([]reading, error)
	CreateReading(context.Context, reading) (reading, error)
	AggregateReadings(context.Context, time.Duration) ([]aggregate, error)
	Close()
}

type postgresStore struct {
	pool    *pgxpool.Pool
	metrics *metrics
}

func newPostgresStore(ctx context.Context, dsn string, m *metrics) (*postgresStore, error) {
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		return nil, fmt.Errorf("create database pool: %w", err)
	}
	return &postgresStore{pool: pool, metrics: m}, nil
}

func (s *postgresStore) observe(ctx context.Context, operation string) (context.Context, func()) {
	ctx, span := otel.Tracer("nereus-api/database").Start(ctx, operation)
	started := time.Now()
	return ctx, func() {
		s.metrics.db.WithLabelValues(operation).Observe(time.Since(started).Seconds())
		span.End()
	}
}

func (s *postgresStore) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }
func (s *postgresStore) Close()                         { s.pool.Close() }

func (s *postgresStore) Migrate(ctx context.Context) error {
	ctx, done := s.observe(ctx, "migrate")
	defer done()
	if _, err := s.pool.Exec(ctx, migrationSQL); err != nil {
		return fmt.Errorf("apply migrations: %w", err)
	}
	return nil
}

func (s *postgresStore) ListBuoys(ctx context.Context) ([]buoy, error) {
	ctx, done := s.observe(ctx, "list_buoys")
	defer done()
	rows, err := s.pool.Query(ctx, `SELECT id, name, latitude, longitude, created_at FROM buoys ORDER BY created_at`)
	if err != nil {
		return nil, fmt.Errorf("query buoys: %w", err)
	}
	defer rows.Close()
	items, err := pgx.CollectRows(rows, pgx.RowToStructByName[buoy])
	if err != nil {
		return nil, fmt.Errorf("collect buoys: %w", err)
	}
	return items, nil
}

func (s *postgresStore) CreateBuoy(ctx context.Context, b buoy) (buoy, error) {
	ctx, done := s.observe(ctx, "create_buoy")
	defer done()
	err := s.pool.QueryRow(ctx, `INSERT INTO buoys (id,name,latitude,longitude) VALUES ($1,$2,$3,$4) RETURNING created_at`, b.ID, b.Name, b.Latitude, b.Longitude).Scan(&b.CreatedAt)
	if err != nil {
		return buoy{}, fmt.Errorf("insert buoy: %w", err)
	}
	return b, nil
}

func (s *postgresStore) GetBuoy(ctx context.Context, id string) (buoy, error) {
	ctx, done := s.observe(ctx, "get_buoy")
	defer done()
	var b buoy
	err := s.pool.QueryRow(ctx, `SELECT id, name, latitude, longitude, created_at FROM buoys WHERE id=$1`, id).Scan(&b.ID, &b.Name, &b.Latitude, &b.Longitude, &b.CreatedAt)
	if err != nil {
		return buoy{}, fmt.Errorf("select buoy: %w", err)
	}
	return b, nil
}

func (s *postgresStore) DeleteBuoy(ctx context.Context, id string) error {
	ctx, done := s.observe(ctx, "delete_buoy")
	defer done()
	tag, err := s.pool.Exec(ctx, `DELETE FROM buoys WHERE id=$1`, id)
	if err != nil {
		return fmt.Errorf("delete buoy: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return pgx.ErrNoRows
	}
	return nil
}

func (s *postgresStore) ListReadings(ctx context.Context, buoyID string, from, to *time.Time, limit, offset int) ([]reading, error) {
	ctx, done := s.observe(ctx, "list_readings")
	defer done()
	rows, err := s.pool.Query(ctx, `SELECT id, buoy_id, water_temp, wave_height, salinity, recorded_at FROM readings WHERE ($1::text='' OR buoy_id=NULLIF($1,'')::uuid) AND ($2::timestamptz IS NULL OR recorded_at >= $2) AND ($3::timestamptz IS NULL OR recorded_at <= $3) ORDER BY recorded_at DESC LIMIT $4 OFFSET $5`, buoyID, from, to, limit, offset)
	if err != nil {
		return nil, fmt.Errorf("query readings: %w", err)
	}
	defer rows.Close()
	items, err := pgx.CollectRows(rows, pgx.RowToStructByName[reading])
	if err != nil {
		return nil, fmt.Errorf("collect readings: %w", err)
	}
	return items, nil
}

func (s *postgresStore) CreateReading(ctx context.Context, r reading) (reading, error) {
	ctx, done := s.observe(ctx, "create_reading")
	defer done()
	err := s.pool.QueryRow(ctx, `INSERT INTO readings (id,buoy_id,water_temp,wave_height,salinity,recorded_at) VALUES ($1,$2,$3,$4,$5,COALESCE($6,now())) RETURNING recorded_at`, r.ID, r.BuoyID, r.WaterTemp, r.WaveHeight, r.Salinity, nullableTime(r.RecordedAt)).Scan(&r.RecordedAt)
	if err != nil {
		return reading{}, fmt.Errorf("insert reading: %w", err)
	}
	return r, nil
}

func nullableTime(value time.Time) any {
	if value.IsZero() {
		return nil
	}
	return value
}

func (s *postgresStore) AggregateReadings(ctx context.Context, window time.Duration) ([]aggregate, error) {
	ctx, done := s.observe(ctx, "aggregate_readings")
	defer done()
	seconds := int64(window.Seconds())
	rows, err := s.pool.Query(ctx, `SELECT to_timestamp(floor(extract(epoch FROM recorded_at)/$1)*$1) AS bucket, avg(water_temp) AS average_water_temp, avg(wave_height) AS average_wave_height, avg(salinity) AS average_salinity, count(*) AS count FROM readings GROUP BY bucket ORDER BY bucket DESC`, seconds)
	if err != nil {
		return nil, fmt.Errorf("aggregate readings: %w", err)
	}
	defer rows.Close()
	items, err := pgx.CollectRows(rows, pgx.RowToStructByName[aggregate])
	if err != nil {
		return nil, fmt.Errorf("collect aggregates: %w", err)
	}
	return items, nil
}