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, adminHumanId 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, humanId string) error
	removeMember(ctx context.Context, networkID, humanId string) error
	getMemberHumanIds(ctx context.Context, networkID string) ([]string, error)
	getNetworksForHuman(ctx context.Context, humanId string) ([]*Network, error)
	isMember(ctx context.Context, networkID, humanId string) (bool, error)

	// Invitations
	createInvitation(ctx context.Context, networkID, email string) error
	getInvitationsByEmail(ctx context.Context, email string) ([]*Invitation, error)
	getInvitationsByNetwork(ctx context.Context, networkID string) ([]*Invitation, error)
	deleteInvitation(ctx context.Context, networkID, email 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, adminHumanId 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_human_id) VALUES ($1, $2, $3)
		 RETURNING id, name, admin_human_id, open_stream_capacity, open_stream_count, created_at`,
		id.String(), name, adminHumanId,
	).Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.OpenStreamCapacity, &n.OpenStreamCount, &n.CreatedAt)
	if err != nil {
		return nil, err
	}

	n.MemberHumanIds = []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_human_id, open_stream_capacity, open_stream_count, created_at FROM networks WHERE id = $1`,
		id,
	).Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.OpenStreamCapacity, &n.OpenStreamCount, &n.CreatedAt)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errNotFound
		}
		return nil, err
	}

	n.MemberHumanIds, err = r.getMemberHumanIds(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, humanId string) error {
	_, err := r.pool.Exec(ctx,
		`INSERT INTO network_members (network_id, human_id) VALUES ($1, $2)
		 ON CONFLICT (network_id, human_id) DO NOTHING`,
		networkID, humanId,
	)
	return err
}

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

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

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

func (r *repositoryImpl) getNetworksForHuman(ctx context.Context, humanId string) ([]*Network, error) {
	rows, err := r.pool.Query(ctx,
		`SELECT n.id, n.name, n.admin_human_id, n.open_stream_capacity, n.open_stream_count, n.created_at
		 FROM networks n
		 WHERE n.admin_human_id = $1
		 OR EXISTS (SELECT 1 FROM network_members nm WHERE nm.network_id = n.id AND nm.human_id = $1)`,
		humanId,
	)
	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.AdminHumanId, &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.MemberHumanIds, err = r.getMemberHumanIds(ctx, n.ID)
		if err != nil {
			return nil, err
		}
	}

	return networks, nil
}

func (r *repositoryImpl) isMember(ctx context.Context, networkID, humanId 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.human_id = $2
			WHERE n.id = $1 AND (n.admin_human_id = $2 OR nm.human_id IS NOT NULL)
		)
	`, networkID, humanId).Scan(&isMember)
	return isMember, err
}

// Invitation methods

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

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

	var invitations []*Invitation
	for rows.Next() {
		var inv Invitation
		if err := rows.Scan(&inv.NetworkID, &inv.Email, &inv.CreatedAt); err != nil {
			return nil, err
		}
		invitations = append(invitations, &inv)
	}
	return invitations, rows.Err()
}

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

	var invitations []*Invitation
	for rows.Next() {
		var inv Invitation
		if err := rows.Scan(&inv.NetworkID, &inv.Email, &inv.CreatedAt); err != nil {
			return nil, err
		}
		invitations = append(invitations, &inv)
	}
	return invitations, rows.Err()
}

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