intake/core/source.go

package core

import (
	"database/sql"
	"errors"
	"fmt"

	_ "github.com/mattn/go-sqlite3"
)

func AddSource(db *DB, name string) error {
	_, err := db.Exec(`
		insert into sources (name)
		values (?)
	`, name)

	return err
}

func GetSources(db *DB) ([]string, error) {
	rows, err := db.Query(`
		select name
		from sources
	`)
	if err != nil {
		return nil, err
	}
	var names []string
	for rows.Next() {
		var name string
		if err = rows.Scan(&name); err != nil {
			return nil, err
		}
		names = append(names, name)
	}
	return names, nil
}

func DeleteSource(db *DB, name string) error {
	_, err := db.Exec(`
		delete from sources
		where name = ?
	`, name)

	return err
}

func AddItem(
	db *DB,
	source string,
	id string,
	title string,
	author string,
	body string,
	link string,
	time int,
) error {
	_, err := db.Exec(`
		insert into items (source, id, active, title, author, body, link, time)
		values (?, ?, ?, ?, ?, ?, ?, ?)
	`, source, id, true, title, author, body, link, time)

	return err
}

func AddItems(db *DB, items []Item) error {
	return db.Transact(func(tx *sql.Tx) error {
		stmt, err := tx.Prepare(`
			insert into items (source, id, active, title, author, body, link, time)
			values (?, ?, ?, ?, ?, ?, ?, ?)
		`)
		if err != nil {
			return err
		}
		for _, item := range items {
			_, err = stmt.Exec(item.Source, item.Id, true, item.Title, item.Author, item.Body, item.Link, item.Time)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

// Deactivate an item, returning its previous active state.
func DeactivateItem(db *DB, source string, id string) (bool, error) {
	row := db.QueryRow(`
		select active
		from items
		where source = ? and id = ?
	`, source, id)
	var active bool
	err := row.Scan(&active)
	if err != nil && errors.Is(err, sql.ErrNoRows) {
		return false, fmt.Errorf("item %s/%s not found", source, id)
	}

	_, err = db.Exec(`
		update items
		set active = 0
		where source = ? and id = ?
	`, source, id)
	if err != nil {
		return false, err
	}
	return active, nil
}

func DeleteItem(db *DB, source string, id string) (int64, error) {
	res, err := db.Exec(`
		delete from items
		where source = ?
		and id = ?
	`, source, id)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

func getItems(db *DB, query string, args ...any) ([]Item, error) {
	rows, err := db.Query(query, args...)
	if err != nil {
		return nil, err
	}
	var items []Item
	for rows.Next() {
		var item Item
		err = rows.Scan(&item.Source, &item.Id, &item.Created, &item.Active, &item.Title, &item.Author, &item.Body, &item.Link, &item.Time)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, nil
}

func GetItem(db *DB, source string, id string) (Item, error) {
	items, err := getItems(db, `
		select source, id, created, active, title, author, body, link, time
		from items
		where source = ?
		and id = ?
	`, source, id)
	if err != nil {
		return Item{}, err
	}
	if len(items) == 0 {
		return Item{}, fmt.Errorf("no item in %s with id %s", source, id)
	}
	return items[0], nil
}

func GetAllActiveItems(db *DB) ([]Item, error) {
	return getItems(db, `
		select
			source, id, created, active, title, author, body, link, time
		from items
		where active <> 0
	`)
}

func GetAllItems(db *DB) ([]Item, error) {
	return getItems(db, `
		select
			source, id, created, active, title, author, body, link, time
		from items
	`)
}

func GetActiveItemsForSource(db *DB, source string) ([]Item, error) {
	return getItems(db, `
		select
			source, id, created, active, title, author, body, link, time
		from items
		where
			source = ?
		and active <> 0
	`, source)
}

func GetAllItemsForSource(db *DB, source string) ([]Item, error) {
	return getItems(db, `
		select
			source, id, created, active, title, author, body, link, time
		from items
		where
			source = ?
	`, source)
}

// Given the results of a fetch, add new items, update existing items, and delete expired items.
func UpdateWithFetchedItems(db *DB, source string, items []Item) (int, int, error) {
	// Get the existing items
	existingItems, err := GetAllItemsForSource(db, source)
	if err != nil {
		return 0, 0, err
	}
	existingIds := map[string]bool{}
	for _, item := range existingItems {
		existingIds[item.Id] = true
	}

	// Split the fetch into adds and updates
	var newItems []Item
	var updatedItems []Item
	for _, item := range items {
		if existingIds[item.Id] {
			updatedItems = append(updatedItems, item)
		} else {
			newItems = append(newItems, item)
		}
	}

	// Bulk insert the new items
	if err = AddItems(db, newItems); err != nil {
		return 0, 0, err
	}

	// Bulk update the existing items
	// TODO

	// Get the list of expired items
	fetchedIds := map[string]bool{}
	for _, item := range items {
		fetchedIds[item.Id] = true
	}
	expiredIds := map[string]bool{}
	for id := range existingIds {
		expiredIds[id] = !fetchedIds[id]
	}

	// Check expired items for deletion
	idsToDelete := map[string]bool{}
	for _, item := range existingItems {
		if expiredIds[item.Id] && item.Deletable() {
			idsToDelete[item.Id] = true
		}
	}

	// Delete each item to be deleted
	for id := range idsToDelete {
		if _, err = DeleteItem(db, source, id); err != nil {
			return 0, 0, err
		}
	}

	return len(newItems), len(idsToDelete), nil
}