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 }