Split source and item functions into different files

core/action.go

@@ -1,18 +1,8 @@
 package core
 
 import (
-	"bufio"
-	"context"
 	"database/sql/driver"
 	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"log"
-	"os"
-	"os/exec"
-	"strings"
-	"time"
 )
 
 // Type alias for storing string array as jsonb
@@ -85,173 +75,3 @@ func DeleteAction(db DB, source string, name string) error {
 	`, source, name)
 	return err
 }
-
-func readStdout(stdout io.ReadCloser, source string, items chan Item, cparse chan bool) {
-	var item Item
-	parseError := false
-	scanout := bufio.NewScanner(stdout)
-	for scanout.Scan() {
-		data := scanout.Bytes()
-		err := json.Unmarshal(data, &item)
-		if err != nil || item.Id == "" {
-			log.Printf("[%s: stdout] %s\n", source, strings.TrimSpace(string(data)))
-			parseError = true
-		} else {
-			item.Active = true   // These fields aren't up to
-			item.Created = 0     // the action to set and
-			item.Source = source // shouldn't be overrideable
-			log.Printf("[%s:   item] %s\n", source, item.Id)
-			items <- item
-		}
-	}
-	// Only send the parsing result at the end, to block main until stdout is drained
-	cparse <- parseError
-	close(items)
-}
-
-func readStderr(stderr io.ReadCloser, source string, done chan bool) {
-	scanerr := bufio.NewScanner(stderr)
-	for scanerr.Scan() {
-		text := strings.TrimSpace(scanerr.Text())
-		log.Printf("[%s: stderr] %s\n", source, text)
-	}
-	done <- true
-}
-
-func writeStdin(stdin io.WriteCloser, text string) {
-	defer stdin.Close()
-	io.WriteString(stdin, text)
-}
-
-func Execute(
-	source string,
-	argv []string,
-	env []string,
-	state []byte,
-	input string,
-	timeout time.Duration,
-) ([]Item, []byte, error) {
-	log.Printf("executing %v", argv)
-
-	if len(argv) == 0 {
-		return nil, nil, errors.New("empty argv")
-	}
-	if source == "" {
-		return nil, nil, errors.New("empty source")
-	}
-
-	stateFile, err := os.CreateTemp("", "intake_state_*")
-	if err != nil {
-		return nil, nil, fmt.Errorf("error: failed to create temp state file: %v", err)
-	}
-	defer func() {
-		if err := os.Remove(stateFile.Name()); err != nil {
-			log.Printf("error: failed to delete %s", stateFile.Name())
-		}
-	}()
-
-	_, err = stateFile.Write(state)
-	if err != nil {
-		return nil, nil, fmt.Errorf("error: failed to write state file: %v", err)
-	}
-
-	env = append(env, "STATE_PATH="+stateFile.Name())
-
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-	cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
-	cmd.Env = append(os.Environ(), env...)
-	cmd.WaitDelay = time.Second * 5
-
-	// Open pipes to the command
-	stdin, err := cmd.StdinPipe()
-	if err != nil {
-		return nil, nil, err
-	}
-	stdout, err := cmd.StdoutPipe()
-	if err != nil {
-		return nil, nil, err
-	}
-	stderr, err := cmd.StderrPipe()
-	if err != nil {
-		return nil, nil, err
-	}
-
-	cout := make(chan Item)
-	cparse := make(chan bool)
-	cerr := make(chan bool)
-
-	// Sink routine for items produced
-	var items []Item
-	go func() {
-		for item := range cout {
-			items = append(items, item)
-		}
-	}()
-
-	// Routines handling the process i/o
-	go writeStdin(stdin, input)
-	go readStdout(stdout, source, cout, cparse)
-	go readStderr(stderr, source, cerr)
-
-	// Kick off the command
-	err = cmd.Start()
-	if err != nil {
-		return nil, nil, err
-	}
-
-	// Block until std{out,err} close
-	<-cerr
-	parseError := <-cparse
-
-	err = cmd.Wait()
-	if ctx.Err() == context.DeadlineExceeded {
-		log.Printf("Timed out after %v\n", timeout)
-		return nil, nil, err
-	} else if exiterr, ok := err.(*exec.ExitError); ok {
-		log.Printf("error: %s failed with exit code %d\n", argv[0], exiterr.ExitCode())
-		return nil, nil, err
-	} else if err != nil {
-		log.Printf("error: %s failed with error: %s\n", argv[0], err)
-		return nil, nil, err
-	}
-
-	if parseError {
-		log.Printf("error: could not parse item\n")
-		return nil, nil, errors.New("invalid JSON")
-	}
-
-	newState, err := os.ReadFile(stateFile.Name())
-	if err != nil {
-		return nil, nil, fmt.Errorf("error: failed to read state file: %v", err)
-	}
-
-	return items, newState, nil
-}
-
-// Execute an action that takes an item as input and returns the item modified.
-// This is basically just a wrapper over [Execute] that handles the input and backfilling.
-func ExecuteItemAction(
-	item Item,
-	argv []string,
-	env []string,
-	state []byte,
-	timeout time.Duration,
-) (Item, []byte, error) {
-	itemJson, err := json.Marshal(item)
-	if err != nil {
-		return Item{}, nil, fmt.Errorf("failed to serialize item: %v", err)
-	}
-
-	res, newState, err := Execute(item.Source, argv, env, state, string(itemJson), timeout)
-	if err != nil {
-		return Item{}, nil, fmt.Errorf("failed to execute action for %s/%s: %v", item.Source, item.Id, err)
-	}
-	if len(res) != 1 {
-		return Item{}, nil, fmt.Errorf("expected action to produce exactly one item, got %d", len(res))
-	}
-	newItem := res[0]
-	BackfillItem(&newItem, &item)
-
-	return newItem, newState, nil
-}

core/action_test.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"testing"
-	"time"
 )
 
 func TestActionCreate(t *testing.T) {
@@ -54,148 +53,3 @@ func TestActionCreate(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-
-func TestExecute(t *testing.T) {
-	assertLen := func(items []Item, length int) {
-		if len(items) != length {
-			t.Fatalf("Expected %d items, got %d", length, len(items))
-		}
-	}
-	assertNil := func(err error) {
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-	assertNotNil := func(err error) {
-		if err == nil {
-			t.Fatal("expected err")
-		}
-	}
-	execute := func(argv []string) ([]Item, error) {
-		item, _, err := Execute("_", argv, nil, nil, "", time.Minute)
-		return item, err
-	}
-
-	res, err := execute([]string{"true"})
-	assertNil(err)
-	assertLen(res, 0)
-
-	// Exit with error code
-	res, err = execute([]string{"false"})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	res, err = execute([]string{"sh", "-c", "exit 22"})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	// Timeout
-	res, _, err = Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond)
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	// Returning items
-	res, err = execute([]string{"jq", "-cn", `{id: "foo"}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "foo" {
-		t.Fatal("jq -cn test failed")
-	}
-
-	// Read from stdin
-	res, _, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute)
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "bar" {
-		t.Fatal("jq -cR test failed")
-	}
-
-	// Set env
-	res, _, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute)
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "baz" {
-		t.Fatal("jq -cn env test failed")
-	}
-
-	// With logging on stderr
-	res, err = execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "box" {
-		t.Fatal("stderr test failed")
-	}
-
-	// Unsupported item field is silently discarded
-	res, err = execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
-	assertNil(err)
-	assertLen(res, 1)
-
-	// Field with incorrect type fails
-	res, err = execute([]string{"jq", "-cn", `{id: ["list"]}`})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	res, err = execute([]string{"jq", "-cn", `{id: null}`})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	// Items with duplicate ids is not a fetch error, but it will fail to update
-	res, err = execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
-	assertNil(err)
-	assertLen(res, 2)
-
-	// Action keys are detected even with empty values
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-	if res[0].Action["goodbye"] != nil {
-		t.Fatal("nonexistent action should key to nil in Action")
-	}
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-
-	// Read state
-	argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
-	res, _, err = Execute("_", argv, nil, []byte("world"), "", time.Minute)
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Title != "Hello world" {
-		t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
-	}
-
-	// Write state
-	argv = []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
-	res, newState, err := Execute("_", argv, nil, nil, "", time.Minute)
-	assertNil(err)
-	assertLen(res, 1)
-	if string(newState) != "Hello world" {
-		t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
-	}
-}

core/execute.go

@@ -0,0 +1,185 @@
+package core
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+)
+
+func readStdout(stdout io.ReadCloser, source string, items chan Item, cparse chan bool) {
+	var item Item
+	parseError := false
+	scanout := bufio.NewScanner(stdout)
+	for scanout.Scan() {
+		data := scanout.Bytes()
+		err := json.Unmarshal(data, &item)
+		if err != nil || item.Id == "" {
+			log.Printf("[%s: stdout] %s\n", source, strings.TrimSpace(string(data)))
+			parseError = true
+		} else {
+			item.Active = true   // These fields aren't up to
+			item.Created = 0     // the action to set and
+			item.Source = source // shouldn't be overrideable
+			log.Printf("[%s:   item] %s\n", source, item.Id)
+			items <- item
+		}
+	}
+	// Only send the parsing result at the end, to block main until stdout is drained
+	cparse <- parseError
+	close(items)
+}
+
+func readStderr(stderr io.ReadCloser, source string, done chan bool) {
+	scanerr := bufio.NewScanner(stderr)
+	for scanerr.Scan() {
+		text := strings.TrimSpace(scanerr.Text())
+		log.Printf("[%s: stderr] %s\n", source, text)
+	}
+	done <- true
+}
+
+func writeStdin(stdin io.WriteCloser, text string) {
+	defer stdin.Close()
+	io.WriteString(stdin, text)
+}
+
+func Execute(
+	source string,
+	argv []string,
+	env []string,
+	state []byte,
+	input string,
+	timeout time.Duration,
+) ([]Item, []byte, error) {
+	log.Printf("executing %v", argv)
+
+	if len(argv) == 0 {
+		return nil, nil, errors.New("empty argv")
+	}
+	if source == "" {
+		return nil, nil, errors.New("empty source")
+	}
+
+	stateFile, err := os.CreateTemp("", "intake_state_*")
+	if err != nil {
+		return nil, nil, fmt.Errorf("error: failed to create temp state file: %v", err)
+	}
+	defer func() {
+		if err := os.Remove(stateFile.Name()); err != nil {
+			log.Printf("error: failed to delete %s", stateFile.Name())
+		}
+	}()
+
+	_, err = stateFile.Write(state)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error: failed to write state file: %v", err)
+	}
+
+	env = append(env, "STATE_PATH="+stateFile.Name())
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
+	cmd.Env = append(os.Environ(), env...)
+	cmd.WaitDelay = time.Second * 5
+
+	// Open pipes to the command
+	stdin, err := cmd.StdinPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	cout := make(chan Item)
+	cparse := make(chan bool)
+	cerr := make(chan bool)
+
+	// Sink routine for items produced
+	var items []Item
+	go func() {
+		for item := range cout {
+			items = append(items, item)
+		}
+	}()
+
+	// Routines handling the process i/o
+	go writeStdin(stdin, input)
+	go readStdout(stdout, source, cout, cparse)
+	go readStderr(stderr, source, cerr)
+
+	// Kick off the command
+	err = cmd.Start()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Block until std{out,err} close
+	<-cerr
+	parseError := <-cparse
+
+	err = cmd.Wait()
+	if ctx.Err() == context.DeadlineExceeded {
+		log.Printf("Timed out after %v\n", timeout)
+		return nil, nil, err
+	} else if exiterr, ok := err.(*exec.ExitError); ok {
+		log.Printf("error: %s failed with exit code %d\n", argv[0], exiterr.ExitCode())
+		return nil, nil, err
+	} else if err != nil {
+		log.Printf("error: %s failed with error: %s\n", argv[0], err)
+		return nil, nil, err
+	}
+
+	if parseError {
+		log.Printf("error: could not parse item\n")
+		return nil, nil, errors.New("invalid JSON")
+	}
+
+	newState, err := os.ReadFile(stateFile.Name())
+	if err != nil {
+		return nil, nil, fmt.Errorf("error: failed to read state file: %v", err)
+	}
+
+	return items, newState, nil
+}
+
+// Execute an action that takes an item as input and returns the item modified.
+// This is basically just a wrapper over [Execute] that handles the input and backfilling.
+func ExecuteItemAction(
+	item Item,
+	argv []string,
+	env []string,
+	state []byte,
+	timeout time.Duration,
+) (Item, []byte, error) {
+	itemJson, err := json.Marshal(item)
+	if err != nil {
+		return Item{}, nil, fmt.Errorf("failed to serialize item: %v", err)
+	}
+
+	res, newState, err := Execute(item.Source, argv, env, state, string(itemJson), timeout)
+	if err != nil {
+		return Item{}, nil, fmt.Errorf("failed to execute action for %s/%s: %v", item.Source, item.Id, err)
+	}
+	if len(res) != 1 {
+		return Item{}, nil, fmt.Errorf("expected action to produce exactly one item, got %d", len(res))
+	}
+	newItem := res[0]
+	BackfillItem(&newItem, &item)
+
+	return newItem, newState, nil
+}

core/execute_test.go

@@ -0,0 +1,151 @@
+package core
+
+import (
+	"testing"
+	"time"
+)
+
+func TestExecute(t *testing.T) {
+	assertLen := func(items []Item, length int) {
+		if len(items) != length {
+			t.Fatalf("Expected %d items, got %d", length, len(items))
+		}
+	}
+	assertNil := func(err error) {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	assertNotNil := func(err error) {
+		if err == nil {
+			t.Fatal("expected err")
+		}
+	}
+	execute := func(argv []string) ([]Item, error) {
+		item, _, err := Execute("_", argv, nil, nil, "", time.Minute)
+		return item, err
+	}
+
+	res, err := execute([]string{"true"})
+	assertNil(err)
+	assertLen(res, 0)
+
+	// Exit with error code
+	res, err = execute([]string{"false"})
+	assertNotNil(err)
+	assertLen(res, 0)
+
+	res, err = execute([]string{"sh", "-c", "exit 22"})
+	assertNotNil(err)
+	assertLen(res, 0)
+
+	// Timeout
+	res, _, err = Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond)
+	assertNotNil(err)
+	assertLen(res, 0)
+
+	// Returning items
+	res, err = execute([]string{"jq", "-cn", `{id: "foo"}`})
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Id != "foo" {
+		t.Fatal("jq -cn test failed")
+	}
+
+	// Read from stdin
+	res, _, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute)
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Id != "bar" {
+		t.Fatal("jq -cR test failed")
+	}
+
+	// Set env
+	res, _, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute)
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Id != "baz" {
+		t.Fatal("jq -cn env test failed")
+	}
+
+	// With logging on stderr
+	res, err = execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Id != "box" {
+		t.Fatal("stderr test failed")
+	}
+
+	// Unsupported item field is silently discarded
+	res, err = execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
+	assertNil(err)
+	assertLen(res, 1)
+
+	// Field with incorrect type fails
+	res, err = execute([]string{"jq", "-cn", `{id: ["list"]}`})
+	assertNotNil(err)
+	assertLen(res, 0)
+
+	res, err = execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
+	assertNotNil(err)
+	assertLen(res, 0)
+
+	res, err = execute([]string{"jq", "-cn", `{id: null}`})
+	assertNotNil(err)
+	assertLen(res, 0)
+
+	// Items with duplicate ids is not a fetch error, but it will fail to update
+	res, err = execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
+	assertNil(err)
+	assertLen(res, 2)
+
+	// Action keys are detected even with empty values
+	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Action["hello"] == nil {
+		t.Fatal("missing hello action")
+	}
+	if res[0].Action["goodbye"] != nil {
+		t.Fatal("nonexistent action should key to nil in Action")
+	}
+
+	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Action["hello"] == nil {
+		t.Fatal("missing hello action")
+	}
+
+	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Action["hello"] == nil {
+		t.Fatal("missing hello action")
+	}
+
+	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Action["hello"] == nil {
+		t.Fatal("missing hello action")
+	}
+
+	// Read state
+	argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
+	res, _, err = Execute("_", argv, nil, []byte("world"), "", time.Minute)
+	assertNil(err)
+	assertLen(res, 1)
+	if res[0].Title != "Hello world" {
+		t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
+	}
+
+	// Write state
+	argv = []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
+	res, newState, err := Execute("_", argv, nil, nil, "", time.Minute)
+	assertNil(err)
+	assertLen(res, 1)
+	if string(newState) != "Hello world" {
+		t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
+	}
+}

core/items.go

@@ -0,0 +1,200 @@
+package core
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func AddItems(db DB, items []Item) error {
+	return db.Transact(func(tx DB) error {
+		stmt, err := tx.Prepare(`
+			insert into items (source, id, active, title, author, body, link, time, action)
+			values (?, ?, ?, ?, ?, ?, ?, ?, jsonb(?))
+		`)
+		if err != nil {
+			return fmt.Errorf("failed to prepare insert: %v", err)
+		}
+		for _, item := range items {
+			actions, err := json.Marshal(item.Action)
+			if err != nil {
+				return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
+			}
+			_, err = stmt.Exec(item.Source, item.Id, true, item.Title, item.Author, item.Body, item.Link, item.Time, actions)
+			if err != nil {
+				return fmt.Errorf("failed to insert %s/%s: %v", item.Source, item.Id, err)
+			}
+		}
+		return nil
+	})
+}
+
+// Set fields in the new item to match the old item where the new item's fields are zero-valued.
+// This allows sources to omit fields and let an action set them without a later fetch overwriting
+// the value from the action, e.g. an on-create action archiving a page and setting the link to
+// point to the archive.
+func BackfillItem(new *Item, old *Item) {
+	new.Active = old.Active
+	new.Created = old.Created
+	if new.Author == "" {
+		new.Author = old.Author
+	}
+	if new.Body == "" {
+		new.Body = old.Body
+	}
+	if new.Link == "" {
+		new.Link = old.Link
+	}
+	if new.Time == 0 {
+		new.Time = old.Time
+	}
+	if new.Title == "" {
+		new.Title = old.Title
+	}
+}
+
+func UpdateItems(db DB, items []Item) error {
+	return db.Transact(func(tx DB) error {
+		stmt, err := tx.Prepare(`
+			update items
+			set
+				title = ?,
+				author = ?,
+				body = ?,
+				link = ?,
+				time = ?,
+				action = jsonb(?)
+			where source = ?
+			and id = ?
+		`)
+		if err != nil {
+			return err
+		}
+		for _, item := range items {
+			actions, err := json.Marshal(item.Action)
+			if err != nil {
+				return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
+			}
+			_, err = stmt.Exec(item.Title, item.Author, item.Body, item.Link, item.Time, actions, item.Source, item.Id)
+			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, &item.Action)
+		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, json(action)
+		from items
+		where source = ?
+		and id = ?
+		order by case when time = 0 then created else time end, 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, json(action)
+		from items
+		where active <> 0
+		order by case when time = 0 then created else time end, id
+	`)
+}
+
+func GetAllItems(db DB) ([]Item, error) {
+	return getItems(db, `
+		select
+			source, id, created, active, title, author, body, link, time, json(action)
+		from items
+		order by case when time = 0 then created else time end, id
+	`)
+}
+
+func GetActiveItemsForSource(db DB, source string) ([]Item, error) {
+	return getItems(db, `
+		select
+			source, id, created, active, title, author, body, link, time, json(action)
+		from items
+		where
+			source = ?
+		and active <> 0
+		order by case when time = 0 then created else time end, id
+	`, source)
+}
+
+func GetAllItemsForSource(db DB, source string) ([]Item, error) {
+	return getItems(db, `
+		select
+			source, id, created, active, title, author, body, link, time, json(action)
+		from items
+		where
+			source = ?
+		order by case when time = 0 then created else time end, id
+	`, source)
+}

core/items_test.go

@@ -0,0 +1,92 @@
+package core
+
+import (
+	"fmt"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func AssertItemIs(t *testing.T, item Item, expected string) {
+	actual := fmt.Sprintf(
+		"%s/%s/%t/%s/%s/%s/%s/%d",
+		item.Source,
+		item.Id,
+		item.Active,
+		item.Title,
+		item.Author,
+		item.Body,
+		item.Link,
+		item.Time,
+	)
+	if actual != expected {
+		t.Fatalf("expected %s, got %s", expected, actual)
+	}
+}
+
+func TestAddItem(t *testing.T) {
+	db := EphemeralDb(t)
+	if err := AddSource(db, "test"); err != nil {
+		t.Fatalf("failed to add source: %v", err)
+	}
+
+	if err := AddItems(db, []Item{
+		{"test", "one", 0, true, "", "", "", "", 0, nil},
+		{"test", "two", 0, true, "title", "author", "body", "link", 123456, nil},
+	}); err != nil {
+		t.Fatalf("failed to add items: %v", err)
+	}
+	items, err := GetActiveItemsForSource(db, "test")
+	if err != nil {
+		t.Fatalf("failed to get active items: %v", err)
+	}
+	if len(items) != 2 {
+		t.Fatal("should get two items")
+	}
+	// order is by (time ?? created) so this ordering is correct as long as you don't run it in early 1970
+	AssertItemIs(t, items[0], "test/two/true/title/author/body/link/123456")
+	AssertItemIs(t, items[1], "test/one/true/////0")
+
+	if _, err = DeactivateItem(db, "test", "one"); err != nil {
+		t.Fatal(err)
+	}
+	items, err = GetActiveItemsForSource(db, "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(items) != 1 {
+		t.Fatal("should get one item")
+	}
+
+	items, err = GetAllItemsForSource(db, "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(items) != 2 {
+		t.Fatal("should get two items")
+	}
+
+	deleted, err := DeleteItem(db, "test", "one")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if deleted != 1 {
+		t.Fatal("expected one deletion")
+	}
+
+	deleted, err = DeleteItem(db, "test", "one")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if deleted != 0 {
+		t.Fatal("expected no deletion")
+	}
+
+	items, err = GetAllItemsForSource(db, "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(items) != 1 {
+		t.Fatal("should get one item")
+	}
+}

core/source.go

@@ -1,9 +1,6 @@
 package core
 
 import (
-	"database/sql"
-	"encoding/json"
-	"errors"
 	"fmt"
 	"log"
 	"time"
@@ -55,196 +52,6 @@ func DeleteSource(db DB, name string) error {
 	return err
 }
 
-func AddItems(db DB, items []Item) error {
-	return db.Transact(func(tx DB) error {
-		stmt, err := tx.Prepare(`
-			insert into items (source, id, active, title, author, body, link, time, action)
-			values (?, ?, ?, ?, ?, ?, ?, ?, jsonb(?))
-		`)
-		if err != nil {
-			return fmt.Errorf("failed to prepare insert: %v", err)
-		}
-		for _, item := range items {
-			actions, err := json.Marshal(item.Action)
-			if err != nil {
-				return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
-			}
-			_, err = stmt.Exec(item.Source, item.Id, true, item.Title, item.Author, item.Body, item.Link, item.Time, actions)
-			if err != nil {
-				return fmt.Errorf("failed to insert %s/%s: %v", item.Source, item.Id, err)
-			}
-		}
-		return nil
-	})
-}
-
-// Set fields in the new item to match the old item where the new item's fields are zero-valued.
-// This allows sources to omit fields and let an action set them without a later fetch overwriting
-// the value from the action, e.g. an on-create action archiving a page and setting the link to
-// point to the archive.
-func BackfillItem(new *Item, old *Item) {
-	new.Active = old.Active
-	new.Created = old.Created
-	if new.Author == "" {
-		new.Author = old.Author
-	}
-	if new.Body == "" {
-		new.Body = old.Body
-	}
-	if new.Link == "" {
-		new.Link = old.Link
-	}
-	if new.Time == 0 {
-		new.Time = old.Time
-	}
-	if new.Title == "" {
-		new.Title = old.Title
-	}
-}
-
-func UpdateItems(db DB, items []Item) error {
-	return db.Transact(func(tx DB) error {
-		stmt, err := tx.Prepare(`
-			update items
-			set
-				title = ?,
-				author = ?,
-				body = ?,
-				link = ?,
-				time = ?,
-				action = jsonb(?)
-			where source = ?
-			and id = ?
-		`)
-		if err != nil {
-			return err
-		}
-		for _, item := range items {
-			actions, err := json.Marshal(item.Action)
-			if err != nil {
-				return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
-			}
-			_, err = stmt.Exec(item.Title, item.Author, item.Body, item.Link, item.Time, actions, item.Source, item.Id)
-			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, &item.Action)
-		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, json(action)
-		from items
-		where source = ?
-		and id = ?
-		order by case when time = 0 then created else time end, 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, json(action)
-		from items
-		where active <> 0
-		order by case when time = 0 then created else time end, id
-	`)
-}
-
-func GetAllItems(db DB) ([]Item, error) {
-	return getItems(db, `
-		select
-			source, id, created, active, title, author, body, link, time, json(action)
-		from items
-		order by case when time = 0 then created else time end, id
-	`)
-}
-
-func GetActiveItemsForSource(db DB, source string) ([]Item, error) {
-	return getItems(db, `
-		select
-			source, id, created, active, title, author, body, link, time, json(action)
-		from items
-		where
-			source = ?
-		and active <> 0
-		order by case when time = 0 then created else time end, id
-	`, source)
-}
-
-func GetAllItemsForSource(db DB, source string) ([]Item, error) {
-	return getItems(db, `
-		select
-			source, id, created, active, title, author, body, link, time, json(action)
-		from items
-		where
-			source = ?
-		order by case when time = 0 then created else time end, id
-	`, source)
-}
-
 func GetState(db DB, source string) ([]byte, error) {
 	row := db.QueryRow("select state from sources where name = ?", source)
 	var state []byte

core/source_test.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"errors"
-	"fmt"
 	"slices"
 	"strings"
 	"testing"
@@ -46,90 +45,6 @@ func TestCreateSource(t *testing.T) {
 	}
 }
 
-func AssertItemIs(t *testing.T, item Item, expected string) {
-	actual := fmt.Sprintf(
-		"%s/%s/%t/%s/%s/%s/%s/%d",
-		item.Source,
-		item.Id,
-		item.Active,
-		item.Title,
-		item.Author,
-		item.Body,
-		item.Link,
-		item.Time,
-	)
-	if actual != expected {
-		t.Fatalf("expected %s, got %s", expected, actual)
-	}
-}
-
-func TestAddItem(t *testing.T) {
-	db := EphemeralDb(t)
-	if err := AddSource(db, "test"); err != nil {
-		t.Fatalf("failed to add source: %v", err)
-	}
-
-	if err := AddItems(db, []Item{
-		{"test", "one", 0, true, "", "", "", "", 0, nil},
-		{"test", "two", 0, true, "title", "author", "body", "link", 123456, nil},
-	}); err != nil {
-		t.Fatalf("failed to add items: %v", err)
-	}
-	items, err := GetActiveItemsForSource(db, "test")
-	if err != nil {
-		t.Fatalf("failed to get active items: %v", err)
-	}
-	if len(items) != 2 {
-		t.Fatal("should get two items")
-	}
-	// order is by (time ?? created) so this ordering is correct as long as you don't run it in early 1970
-	AssertItemIs(t, items[0], "test/two/true/title/author/body/link/123456")
-	AssertItemIs(t, items[1], "test/one/true/////0")
-
-	if _, err = DeactivateItem(db, "test", "one"); err != nil {
-		t.Fatal(err)
-	}
-	items, err = GetActiveItemsForSource(db, "test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(items) != 1 {
-		t.Fatal("should get one item")
-	}
-
-	items, err = GetAllItemsForSource(db, "test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(items) != 2 {
-		t.Fatal("should get two items")
-	}
-
-	deleted, err := DeleteItem(db, "test", "one")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if deleted != 1 {
-		t.Fatal("expected one deletion")
-	}
-
-	deleted, err = DeleteItem(db, "test", "one")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if deleted != 0 {
-		t.Fatal("expected no deletion")
-	}
-
-	items, err = GetAllItemsForSource(db, "test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(items) != 1 {
-		t.Fatal("should get one item")
-	}
-}
-
 func TestUpdateSourceAddAndDelete(t *testing.T) {
 	db := EphemeralDb(t)
 	if err := AddSource(db, "test"); err != nil {