mirror of
https://github.com/ThisTine/Snip.git
synced 2026-08-19 07:28:46 +07:00
feat: first commit
This commit is contained in:
sittichok Ouamsiri
@@ -0,0 +1,98 @@
|
||||
// Package postgres implements the repository ports on top of pgx/pgxpool.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
const schema = `
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id TEXT PRIMARY KEY,
|
||||
email TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
provider TEXT NOT NULL,
|
||||
subject TEXT NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
UNIQUE (provider, subject)
|
||||
);
|
||||
|
||||
CREATE SEQUENCE IF NOT EXISTS link_code_seq;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS links (
|
||||
id TEXT PRIMARY KEY,
|
||||
code TEXT UNIQUE NOT NULL,
|
||||
long_url TEXT NOT NULL,
|
||||
mode TEXT NOT NULL,
|
||||
has_pin BOOLEAN NOT NULL DEFAULT false,
|
||||
pin_hash TEXT NOT NULL DEFAULT '',
|
||||
owner_id TEXT,
|
||||
total_clicks BIGINT NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS links_owner_idx ON links (owner_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS click_daily (
|
||||
code TEXT NOT NULL,
|
||||
day DATE NOT NULL,
|
||||
count BIGINT NOT NULL DEFAULT 0,
|
||||
PRIMARY KEY (code, day)
|
||||
);
|
||||
`
|
||||
|
||||
// migrationLockKey is an arbitrary, stable key for the advisory lock that
|
||||
// serializes schema creation across concurrently-starting services.
|
||||
const migrationLockKey = 947213
|
||||
|
||||
// Migrate opens a short-lived connection pool, runs the schema migrations, and
|
||||
// closes the pool. Safe to call concurrently: an advisory lock serialises DDL
|
||||
// across multiple callers. All statements use IF NOT EXISTS so re-runs are
|
||||
// idempotent and no data is lost.
|
||||
func Migrate(ctx context.Context, dsn string) error {
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer pool.Close()
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return migrate(ctx, pool)
|
||||
}
|
||||
|
||||
// Connect opens a pool and applies the schema (idempotent migrations). Because
|
||||
// all three commands may boot at once and each runs migrations, the DDL is
|
||||
// guarded by a session advisory lock so concurrent `CREATE ... IF NOT EXISTS`
|
||||
// can't race on the Postgres catalog.
|
||||
func Connect(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
pool.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := migrate(ctx, pool); err != nil {
|
||||
pool.Close()
|
||||
return nil, err
|
||||
}
|
||||
return pool, nil
|
||||
}
|
||||
|
||||
func migrate(ctx context.Context, pool *pgxpool.Pool) error {
|
||||
conn, err := pool.Acquire(ctx) // advisory lock is session-scoped → one conn
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Release()
|
||||
|
||||
if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", migrationLockKey); err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Exec(ctx, "SELECT pg_advisory_unlock($1)", migrationLockKey)
|
||||
|
||||
_, err = conn.Exec(ctx, schema)
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,230 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/snip/backend/internal/domain"
|
||||
)
|
||||
|
||||
type LinkRepo struct{ pool *pgxpool.Pool }
|
||||
|
||||
func NewLinkRepo(pool *pgxpool.Pool) *LinkRepo { return &LinkRepo{pool: pool} }
|
||||
|
||||
func isUniqueViolation(err error) bool {
|
||||
var pgErr *pgconn.PgError
|
||||
return errors.As(err, &pgErr) && pgErr.Code == "23505"
|
||||
}
|
||||
|
||||
func (r *LinkRepo) Create(ctx context.Context, l *domain.Link) error {
|
||||
_, err := r.pool.Exec(ctx,
|
||||
`INSERT INTO links (id, code, long_url, mode, has_pin, pin_hash, owner_id, total_clicks, created_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,NULLIF($7,''),0,$8)`,
|
||||
l.ID, l.Code, l.LongURL, string(l.Mode), l.HasPin, l.PinHash, l.OwnerID, l.CreatedAt)
|
||||
if isUniqueViolation(err) {
|
||||
return domain.ErrCodeTaken
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
const selectCols = `id, code, long_url, mode, has_pin, pin_hash, COALESCE(owner_id,''), total_clicks, created_at`
|
||||
|
||||
func scanLink(row pgx.Row) (*domain.Link, error) {
|
||||
var l domain.Link
|
||||
var mode string
|
||||
if err := row.Scan(&l.ID, &l.Code, &l.LongURL, &mode, &l.HasPin, &l.PinHash, &l.OwnerID, &l.TotalClicks, &l.CreatedAt); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, domain.ErrNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
l.Mode = domain.Mode(mode)
|
||||
return &l, nil
|
||||
}
|
||||
|
||||
// GetByCode is the redirect hot-path read (cache miss only). It deliberately
|
||||
// skips the 7-day stats join — the redirect doesn't need them.
|
||||
func (r *LinkRepo) GetByCode(ctx context.Context, code string) (*domain.Link, error) {
|
||||
return scanLink(r.pool.QueryRow(ctx, `SELECT `+selectCols+` FROM links WHERE code=$1`, code))
|
||||
}
|
||||
|
||||
func (r *LinkRepo) GetByID(ctx context.Context, id string) (*domain.Link, error) {
|
||||
l, err := scanLink(r.pool.QueryRow(ctx, `SELECT `+selectCols+` FROM links WHERE id=$1`, id))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
weeks, err := r.weeksFor(ctx, []string{l.Code})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
l.Last7Days = fillWeek(weeks[l.Code], time.Now().UTC())
|
||||
return l, nil
|
||||
}
|
||||
|
||||
func (r *LinkRepo) ListByOwner(ctx context.Context, ownerID, query string, limit, offset int) ([]domain.Link, int, error) {
|
||||
// `pattern` is NULL when there's no query, so the filter is skipped.
|
||||
var pattern any
|
||||
if query != "" {
|
||||
pattern = "%" + query + "%"
|
||||
}
|
||||
|
||||
var total int
|
||||
if err := r.pool.QueryRow(ctx,
|
||||
`SELECT COUNT(*) FROM links
|
||||
WHERE owner_id=$1 AND ($2::text IS NULL OR code ILIKE $2 OR long_url ILIKE $2)`,
|
||||
ownerID, pattern).Scan(&total); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if total == 0 {
|
||||
return nil, 0, nil
|
||||
}
|
||||
|
||||
rows, err := r.pool.Query(ctx,
|
||||
`SELECT `+selectCols+` FROM links
|
||||
WHERE owner_id=$1 AND ($2::text IS NULL OR code ILIKE $2 OR long_url ILIKE $2)
|
||||
ORDER BY created_at DESC LIMIT $3 OFFSET $4`,
|
||||
ownerID, pattern, limit, offset)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var links []domain.Link
|
||||
var codes []string
|
||||
for rows.Next() {
|
||||
l, err := scanLink(rows)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
links = append(links, *l)
|
||||
codes = append(codes, l.Code)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
weeks, err := r.weeksFor(ctx, codes)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
for i := range links {
|
||||
links[i].Last7Days = fillWeek(weeks[links[i].Code], time.Now().UTC())
|
||||
}
|
||||
return links, total, nil
|
||||
}
|
||||
|
||||
func (r *LinkRepo) StatsByOwner(ctx context.Context, ownerID string) (domain.OwnerStats, error) {
|
||||
var s domain.OwnerStats
|
||||
if err := r.pool.QueryRow(ctx,
|
||||
`SELECT COUNT(*), COALESCE(SUM(total_clicks),0) FROM links WHERE owner_id=$1`,
|
||||
ownerID).Scan(&s.TotalLinks, &s.TotalClicks); err != nil {
|
||||
return s, err
|
||||
}
|
||||
since := time.Now().UTC().AddDate(0, 0, -6).Format("2006-01-02")
|
||||
if err := r.pool.QueryRow(ctx,
|
||||
`SELECT COALESCE(SUM(cd.count),0) FROM click_daily cd
|
||||
JOIN links l ON l.code = cd.code
|
||||
WHERE l.owner_id=$1 AND cd.day >= $2`,
|
||||
ownerID, since).Scan(&s.WeekClicks); err != nil {
|
||||
return s, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (r *LinkRepo) Update(ctx context.Context, l *domain.Link) error {
|
||||
tag, err := r.pool.Exec(ctx,
|
||||
`UPDATE links SET long_url=$2, has_pin=$3, pin_hash=$4 WHERE id=$1`,
|
||||
l.ID, l.LongURL, l.HasPin, l.PinHash)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return domain.ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *LinkRepo) Delete(ctx context.Context, id, ownerID string) error {
|
||||
tag, err := r.pool.Exec(ctx, `DELETE FROM links WHERE id=$1 AND owner_id=$2`, id, ownerID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return domain.ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *LinkRepo) ExistsCode(ctx context.Context, code string) (bool, error) {
|
||||
var exists bool
|
||||
err := r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM links WHERE code=$1)`, code).Scan(&exists)
|
||||
return exists, err
|
||||
}
|
||||
|
||||
func (r *LinkRepo) NextSequence(ctx context.Context) (int64, error) {
|
||||
var seq int64
|
||||
err := r.pool.QueryRow(ctx, `SELECT nextval('link_code_seq')`).Scan(&seq)
|
||||
return seq, err
|
||||
}
|
||||
|
||||
func (r *LinkRepo) RecordClicks(ctx context.Context, code string, day time.Time, n int64) error {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
if _, err := tx.Exec(ctx,
|
||||
`INSERT INTO click_daily (code, day, count) VALUES ($1,$2,$3)
|
||||
ON CONFLICT (code, day) DO UPDATE SET count = click_daily.count + EXCLUDED.count`,
|
||||
code, day, n); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `UPDATE links SET total_clicks = total_clicks + $2 WHERE code=$1`, code, n); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// weeksFor returns, per code, that code's day->count map for the last 7 days.
|
||||
func (r *LinkRepo) weeksFor(ctx context.Context, codes []string) (map[string]map[string]int64, error) {
|
||||
out := make(map[string]map[string]int64)
|
||||
if len(codes) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
since := time.Now().UTC().AddDate(0, 0, -6).Format("2006-01-02")
|
||||
rows, err := r.pool.Query(ctx,
|
||||
`SELECT code, to_char(day,'YYYY-MM-DD'), count FROM click_daily WHERE code = ANY($1) AND day >= $2`,
|
||||
codes, since)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var code, day string
|
||||
var count int64
|
||||
if err := rows.Scan(&code, &day, &count); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if out[code] == nil {
|
||||
out[code] = make(map[string]int64)
|
||||
}
|
||||
out[code][day] = count
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// fillWeek expands a sparse day->count map into a dense 7-day window.
|
||||
func fillWeek(counts map[string]int64, now time.Time) []domain.DayCount {
|
||||
out := make([]domain.DayCount, 0, 7)
|
||||
for i := 6; i >= 0; i-- {
|
||||
key := now.AddDate(0, 0, -i).Format("2006-01-02")
|
||||
out = append(out, domain.DayCount{Date: key, Count: counts[key]})
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/snip/backend/internal/domain"
|
||||
)
|
||||
|
||||
type UserRepo struct{ pool *pgxpool.Pool }
|
||||
|
||||
func NewUserRepo(pool *pgxpool.Pool) *UserRepo { return &UserRepo{pool: pool} }
|
||||
|
||||
func (r *UserRepo) Upsert(ctx context.Context, u *domain.User) (*domain.User, error) {
|
||||
id := newUserID()
|
||||
var out domain.User
|
||||
err := r.pool.QueryRow(ctx,
|
||||
`INSERT INTO users (id, email, name, provider, subject)
|
||||
VALUES ($1,$2,$3,$4,$5)
|
||||
ON CONFLICT (provider, subject)
|
||||
DO UPDATE SET email = EXCLUDED.email, name = EXCLUDED.name
|
||||
RETURNING id, email, name, provider, subject`,
|
||||
id, u.Email, u.Name, u.Provider, u.Subject).
|
||||
Scan(&out.ID, &out.Email, &out.Name, &out.Provider, &out.Subject)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
func (r *UserRepo) GetByID(ctx context.Context, id string) (*domain.User, error) {
|
||||
var u domain.User
|
||||
err := r.pool.QueryRow(ctx,
|
||||
`SELECT id, email, name, provider, subject FROM users WHERE id=$1`, id).
|
||||
Scan(&u.ID, &u.Email, &u.Name, &u.Provider, &u.Subject)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, domain.ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &u, nil
|
||||
}
|
||||
|
||||
func newUserID() string {
|
||||
const hex = "0123456789abcdef"
|
||||
b := make([]byte, 16)
|
||||
_, _ = rand.Read(b)
|
||||
out := make([]byte, 16)
|
||||
for i := range out {
|
||||
out[i] = hex[int(b[i])%16]
|
||||
}
|
||||
return "u_" + string(out)
|
||||
}
|
||||
Reference in New Issue
Block a user