initial commit

internal/store/pg/store.go

@@ -0,0 +1,154 @@
+package pg
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"git.loyso.art/frx/devsim/internal/entities"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func Dial(ctx context.Context, addr string) (*repository, error) {
+	config, err := pgxpool.ParseConfig(addr)
+	if err != nil {
+		return nil, fmt.Errorf("parsing config: %w", err)
+	}
+
+	pool, err := pgxpool.NewWithConfig(ctx, config)
+	if err != nil {
+		return nil, fmt.Errorf("connecting to db: %w", err)
+	}
+
+	return &repository{db: pool}, nil
+}
+
+type repository struct {
+	db *pgxpool.Pool
+}
+
+func (r *repository) StatsRepository() statsRepository {
+	return statsRepository{
+		db: r.db,
+	}
+}
+
+type statsRepository struct {
+	db *pgxpool.Pool
+}
+
+type deviceStatsDB struct {
+	DeviceID        string
+	IncomingTraffic int
+	OutgoingTraffic int
+	IncomingRPS     int
+	ReadRPS         int
+	WriteRPS        int
+	UpdatedAt       time.Time
+}
+
+func (s deviceStatsDB) asDomain() entities.DeviceStatistics {
+	return entities.DeviceStatistics{
+		ID:                   entities.DeviceID(s.DeviceID),
+		IncomingTrafficBytes: s.IncomingTraffic,
+		OutgoingTrafficBytes: s.OutgoingTraffic,
+		IncomingRPS:          s.IncomingRPS,
+		ReadRPS:              s.ReadRPS,
+		WriteRPS:             s.WriteRPS,
+		UpdatedAt:            s.UpdatedAt,
+	}
+}
+
+func (r statsRepository) Upsert(ctx context.Context, id entities.DeviceID, stats entities.DeviceStatistics) error {
+	const query = `INSERT INTO public.stats (
+			device_id,
+			inc_traffic,
+			out_traffic,
+			inc_rps, 
+			read_rps,
+			write_rps,
+			updated_at
+		) VALUES (
+			$1,
+			$2,
+			$3,
+			$4,
+			$5,
+			$6,
+			$7,
+		) ON CONFLICT(device_id) DO UPDATE SET
+			inc_traffic = EXCLUDED.inc_traffic,
+			out_traffic = EXCLUDED.out_traffic,
+			inc_rps = EXCLUDED.inc_rps,
+			read_rps = EXCLUDED.read_rps,
+			write_rps = EXCLUDED.write_rps,
+			updated_at = EXCLUDED.updated_at
+		RETURNING id;
+	`
+
+	_, err := r.db.Exec(
+		ctx, query,
+		id,
+		stats.IncomingTrafficBytes,
+		stats.OutgoingTrafficBytes,
+		stats.IncomingRPS,
+		stats.ReadRPS,
+		stats.WriteRPS,
+	)
+	if err != nil {
+		return fmt.Errorf("executing query: %w", err)
+	}
+
+	return nil
+}
+
+func (r statsRepository) List(ctx context.Context) (out []entities.DeviceStatistics, err error) {
+	var count int
+	err = r.db.QueryRow(ctx, `SELECT COUNT(device_id) FROM public.stats`).Scan(&count)
+	if err != nil {
+		return nil, fmt.Errorf("getting count: %w", err)
+	}
+
+	out = make([]entities.DeviceStatistics, 0, count)
+
+	const query = `SELECT
+		device_id,
+		inc_traffic,
+		out_traffic,
+		inc_rps, 
+		read_rps,
+		write_rps,
+		updated_at
+	FROM public.stats;
+	`
+
+	rows, err := r.db.Query(ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("querying: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var stat deviceStatsDB
+		err = rows.Scan(
+			&stat.DeviceID,
+			&stat.IncomingTraffic,
+			&stat.OutgoingTraffic,
+			&stat.IncomingRPS,
+			&stat.ReadRPS,
+			&stat.WriteRPS,
+			&stat.UpdatedAt,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("scanning row: %w", err)
+		}
+
+		out = append(out, stat.asDomain())
+	}
+	if err = rows.Err(); err != nil {
+		return nil, fmt.Errorf("checking rows err: %w", err)
+	}
+
+	return out, nil
+}