jules/database/postgres/notifications.go

package postgres

import (
	"context"
	"database/sql"

	"github.com/google/uuid"
	"m8sh.su/d/jules/database"
)

type Notifications struct {
	conn *sql.DB
}

func (n *Notifications) Push(ctx context.Context, notif *database.Notification) error {
	_, err := n.conn.ExecContext(ctx, `
		INSERT INTO notifications (id, user_id, initiator_id, scheduled_at, content, repeat_on)
		VALUES ($1, $2, $3, $4, $5, $6)
	`, notif.ID, notif.UserID, notif.InitiatorID, notif.ScheduledAt, notif.Content, notif.RepeatOn)
	return err
}

func (n *Notifications) Pop(ctx context.Context, limit int) ([]database.Notification, error) {
	rows, err := n.conn.QueryContext(ctx, `
		WITH batch AS (
			SELECT id
			FROM notifications
			WHERE scheduled_at <= NOW()
			ORDER BY scheduled_at
			LIMIT $1
			FOR UPDATE SKIP LOCKED
		)
		DELETE FROM notifications
		WHERE id IN (SELECT id FROM batch)
		RETURNING id, user_id, initiator_id, scheduled_at, content, repeat_on
	`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var notifs []database.Notification
	for rows.Next() {
		var n database.Notification
		if err = rows.Scan(&n.ID, &n.UserID, &n.InitiatorID, &n.ScheduledAt, &n.Content, &n.RepeatOn); err != nil {
			return nil, err
		}
		notifs = append(notifs, n)
	}
	return notifs, rows.Err()
}

func (n *Notifications) List(ctx context.Context, userID uuid.UUID) ([]database.Notification, error) {
	rows, err := n.conn.QueryContext(ctx, `
		SELECT id, user_id, initiator_id, scheduled_at, content, repeat_on
		FROM notifications
		WHERE user_id = $1
		ORDER BY scheduled_at
	`, userID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var notifs []database.Notification
	for rows.Next() {
		var notif database.Notification
		if err = rows.Scan(&notif.ID, &notif.UserID, &notif.InitiatorID, &notif.ScheduledAt, &notif.Content, &notif.RepeatOn); err != nil {
			return nil, err
		}
		notifs = append(notifs, notif)
	}
	return notifs, rows.Err()
}

func (n *Notifications) ListOutgoing(ctx context.Context, initiatorID uuid.UUID) ([]database.Notification, error) {
	rows, err := n.conn.QueryContext(ctx, `
		SELECT id, user_id, initiator_id, scheduled_at, content, repeat_on
		FROM notifications
		WHERE initiator_id = $1 AND user_id != initiator_id
		ORDER BY scheduled_at DESC
	`, initiatorID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var notifs []database.Notification
	for rows.Next() {
		var notif database.Notification
		if err = rows.Scan(&notif.ID, &notif.UserID, &notif.InitiatorID, &notif.ScheduledAt, &notif.Content, &notif.RepeatOn); err != nil {
			return nil, err
		}
		notifs = append(notifs, notif)
	}
	return notifs, rows.Err()
}

func (n *Notifications) Delete(ctx context.Context, id uuid.UUID) error {
	result, err := n.conn.ExecContext(ctx, `
		DELETE FROM notifications WHERE id = $1
	`, id)
	if err != nil {
		return err
	}
	rows, _ := result.RowsAffected()
	if rows == 0 {
		return database.ErrNotFound
	}
	return nil
}