llink/go/internal/network/repository.go

package network

import (
	"context"
	"errors"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"go.jetify.com/typeid"
)

var errNotFound = errors.New("not found")

type networkIDPrefix struct{}

func (networkIDPrefix) Prefix() string { return "net" }

type networkID struct {
	typeid.TypeID[networkIDPrefix]
}

func newNetworkID() (networkID, error) {
	return typeid.New[networkID]()
}

var errCapacityExceeded = errors.New("capacity exceeded")

type repository interface {
	create(ctx context.Context, name, adminEmail string) (*Network, error)
	getByID(ctx context.Context, id string) (*Network, error)
	updateName(ctx context.Context, id, name string) error
	delete(ctx context.Context, id string) error
	addMember(ctx context.Context, networkID, email string) error
	removeMember(ctx context.Context, networkID, email string) error
	getMemberEmails(ctx context.Context, networkID string) ([]string, error)
	getNetworksForEmail(ctx context.Context, email string) ([]*Network, error)
	isMember(ctx context.Context, networkID, email string) (bool, error)
	setOpenStreamCapacity(ctx context.Context, id string, capacity int) error
	incrementOpenStreamCount(ctx context.Context, id string) error
	decrementOpenStreamCount(ctx context.Context, id string) error
}

type repositoryImpl struct {
	pool *pgxpool.Pool
}

func newRepository(pool *pgxpool.Pool) repository {
	return &repositoryImpl{pool: pool}
}

func (r *repositoryImpl) create(ctx context.Context, name, adminEmail string) (*Network, error) {
	id, err := newNetworkID()
	if err != nil {
		return nil, err
	}

	var n Network
	err = r.pool.QueryRow(ctx,
		`INSERT INTO networks (id, name, admin_email) VALUES ($1, $2, $3)
		 RETURNING id, name, admin_email, open_stream_capacity, open_stream_count, created_at`,
		id.String(), name, adminEmail,
	).Scan(&n.ID, &n.Name, &n.AdminEmail, &n.OpenStreamCapacity, &n.OpenStreamCount, &n.CreatedAt)
	if err != nil {
		return nil, err
	}

	n.MemberEmails = []string{}
	return &n, nil
}

func (r *repositoryImpl) getByID(ctx context.Context, id string) (*Network, error) {
	var n Network
	err := r.pool.QueryRow(ctx,
		`SELECT id, name, admin_email, open_stream_capacity, open_stream_count, created_at FROM networks WHERE id = $1`,
		id,
	).Scan(&n.ID, &n.Name, &n.AdminEmail, &n.OpenStreamCapacity, &n.OpenStreamCount, &n.CreatedAt)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errNotFound
		}
		return nil, err
	}

	n.MemberEmails, err = r.getMemberEmails(ctx, id)
	if err != nil {
		return nil, err
	}

	return &n, nil
}

func (r *repositoryImpl) updateName(ctx context.Context, id, name string) error {
	result, err := r.pool.Exec(ctx,
		`UPDATE networks SET name = $1 WHERE id = $2`,
		name, id,
	)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return errNotFound
	}
	return nil
}

func (r *repositoryImpl) delete(ctx context.Context, id string) error {
	result, err := r.pool.Exec(ctx, `DELETE FROM networks WHERE id = $1`, id)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return errNotFound
	}
	return nil
}

func (r *repositoryImpl) addMember(ctx context.Context, networkID, email string) error {
	_, err := r.pool.Exec(ctx,
		`INSERT INTO network_members (network_id, email) VALUES ($1, $2)
		 ON CONFLICT (network_id, email) DO NOTHING`,
		networkID, email,
	)
	return err
}

func (r *repositoryImpl) removeMember(ctx context.Context, networkID, email string) error {
	_, err := r.pool.Exec(ctx,
		`DELETE FROM network_members WHERE network_id = $1 AND email = $2`,
		networkID, email,
	)
	return err
}

func (r *repositoryImpl) getMemberEmails(ctx context.Context, networkID string) ([]string, error) {
	rows, err := r.pool.Query(ctx,
		`SELECT email FROM network_members WHERE network_id = $1`,
		networkID,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var emails []string
	for rows.Next() {
		var email string
		if err := rows.Scan(&email); err != nil {
			return nil, err
		}
		emails = append(emails, email)
	}
	return emails, rows.Err()
}

func (r *repositoryImpl) getNetworksForEmail(ctx context.Context, email string) ([]*Network, error) {
	rows, err := r.pool.Query(ctx,
		`SELECT n.id, n.name, n.admin_email, n.open_stream_capacity, n.open_stream_count, n.created_at
		 FROM networks n
		 WHERE n.admin_email = $1
		 OR EXISTS (SELECT 1 FROM network_members nm WHERE nm.network_id = n.id AND nm.email = $1)`,
		email,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var networks []*Network
	for rows.Next() {
		var n Network
		if err := rows.Scan(&n.ID, &n.Name, &n.AdminEmail, &n.OpenStreamCapacity, &n.OpenStreamCount, &n.CreatedAt); err != nil {
			return nil, err
		}
		networks = append(networks, &n)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	for _, n := range networks {
		n.MemberEmails, err = r.getMemberEmails(ctx, n.ID)
		if err != nil {
			return nil, err
		}
	}

	return networks, nil
}

func (r *repositoryImpl) isMember(ctx context.Context, networkID, email string) (bool, error) {
	var isMember bool
	err := r.pool.QueryRow(ctx, `
		SELECT EXISTS(
			SELECT 1 FROM networks n
			LEFT JOIN network_members nm ON nm.network_id = n.id AND nm.email = $2
			WHERE n.id = $1 AND (n.admin_email = $2 OR nm.email IS NOT NULL)
		)
	`, networkID, email).Scan(&isMember)
	return isMember, err
}

func (r *repositoryImpl) setOpenStreamCapacity(ctx context.Context, id string, capacity int) error {
	result, err := r.pool.Exec(ctx,
		`UPDATE networks SET open_stream_capacity = $1 WHERE id = $2`,
		capacity, id,
	)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return errNotFound
	}
	return nil
}

func (r *repositoryImpl) incrementOpenStreamCount(ctx context.Context, id string) error {
	result, err := r.pool.Exec(ctx,
		`UPDATE networks SET open_stream_count = open_stream_count + 1
		 WHERE id = $1 AND open_stream_count < open_stream_capacity`,
		id,
	)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		// Check if network exists vs capacity exceeded
		var exists bool
		err := r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM networks WHERE id = $1)`, id).Scan(&exists)
		if err != nil {
			return err
		}
		if !exists {
			return errNotFound
		}
		return errCapacityExceeded
	}
	return nil
}

func (r *repositoryImpl) decrementOpenStreamCount(ctx context.Context, id string) error {
	result, err := r.pool.Exec(ctx,
		`UPDATE networks SET open_stream_count = GREATEST(0, open_stream_count - 1) WHERE id = $1`,
		id,
	)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return errNotFound
	}
	return nil
}