Update todo list

Using == won't work when the Action field is a map[string]RawMessage

.envrc

@@ -0,0 +1,2 @@
+layout go
+use flake

.gitignore

@@ -0,0 +1,2 @@
+.direnv
+tmp/

Makefile

@@ -0,0 +1,10 @@
+.PHONY: help serve test-data
+
+help: ## display this help
+	@awk 'BEGIN{FS = ":.*##"; printf "\033[1m\nUsage\n \033[1;92m make\033[0;36m <target>\033[0m\n"} /^[a-zA-Z0-9_-]+:.*?##/ { printf "  \033[36m%-15s\033[0m %s\n", $$1, $$2 } ' $(MAKEFILE_LIST)
+
+serve: ## Run "intake serve" with live reload
+	@air -build.cmd "go build -o tmp/intake" -build.bin tmp/intake -build.args_bin serve,--data-dir,tmp
+
+test-data: ## Recreate test data in tmp/
+	@test/test_items.sh

README.md

@@ -0,0 +1,140 @@
+# intake
+
+Intake is an arbitrary feed aggregator that generalizes the concept of a feed.
+Rather than being restricted to parsing items out of an RSS feed, Intake provides a middle layer of executing arbitrary commands that conform to a JSON-based specification.
+An Intake source can parse an RSS feed, but it can also scrape a website without a feed, provide additional logic to filter or annotate feed items, or integrate with an API.
+
+## Development
+
+Parity with existing Python version
+
+* [x] create sources
+* [ ] rename sources
+* fetch sources
+  * [x] create and delete items
+  * [x] update existing items
+  * [ ] support item TTL and TTD
+  * [x] on_create triggers
+  * [ ] on_delete triggers
+  * [x] dry-run
+* item actions
+  * [x] create
+  * [x] edit
+  * [ ] rename
+  * [x] delete
+  * [x] execute
+  * [x] require items to declare action support
+  * [ ] state files
+  * [ ] source environment
+  * [ ] working directory set
+* [ ] update web UI credentials
+* [ ] automatic crontab integration
+* [ ] feed supports item TTS
+* [x] data directory from envvars
+* [ ] source-level tt{s,d,l}
+* [ ] source batching
+* channels
+  * [ ] create
+  * [ ] edit
+  * [ ] rename
+  * [ ] delete
+* feeds
+  * [x] show items
+  * [x] deactivate items
+  * [x] mass deactivate
+  * [ ] punt
+  * [x] trigger actions
+  * [x] add ad-hoc items
+  * [ ] show/hide deactivated items
+  * [ ] show/hide tts items
+  * [x] sort by time ?? created
+  * [ ] paging
+* [ ] NixOS module
+* [ ] NixOS module demo
+
+Additional features
+
+* [ ] metric reporting
+* [ ] on action failure, create an error item with logs
+* [ ] first-party password handling instead of basic auth and htpasswd
+* [ ] items gracefully add new fields and `action` keys
+* [ ] arbitrary date punt
+* [ ] HTTP edit item
+* [ ] sort crontab entries
+* [ ] TUI feed view
+
+## Overview
+
+In Intake, a _source_ represents a single content feed of discrete _items_, such as a blog and its posts or a website and its pages.
+Each source has associated _actions_, which are executable commands.
+The `fetch` action checks the feed and returns the items in a JSON format.
+Each item returned by a fetch is stored by Intake and appears in that feed's source.
+When you have read an item, you can deactivate it, which hides it from your feed.
+When a deactivated item is no longer returned by `fetch`, it is deleted.
+This allows you to consume feed content at your own pace without missing anything.
+
+Intake stores all its data in a SQLite database.
+This database is stored in `$INTAKE_DATA_DIR`, `$XDG_DATA_HOME/intake`, or `$HOME/.local/share/intake`, whichever is resolved first.
+The database can also be specified on the command line via `--data-dir`/`-d` instead of the environment.
+
+### Items
+
+Items are passed between Intake and sources as JSON objects.
+Only the `id` field is required.
+Any unspecified field is equivalent to the empty string, object, or 0, depending on field's type.
+
+| Field name | Specification | Description |
+| ---------- | ------------- | ----------- |
+| `id`       | **Required**  | A unique identifier within the source.
+| `source`   | **Automatic** | The source that produced the item.
+| `created`  | **Automatic** | The Unix timestamp at which Intake first processed the item.
+| `active`   | **Automatic** | Whether the item is active and displayed in feeds.
+| `title`    | Optional      | The title of the item. If an item has no title, `id` is used as a fallback title.
+| `author`   | Optional      | An author name associated with the item. Displayed in the item footer.
+| `body`     | Optional      | Body text of the item as raw HTML. This will be displayed in the item without further processing! Consider your sources' threat models against injection attacks.
+| `link`     | Optional      | A hyperlink associated with the item.
+| `time`     | Optional      | A Unix timestamp associated with the item, not necessarily when the item was created. Items sort by `time` when it is defined and fall back to `created`. Displayed in the item footer.
+| `action`   | Optional      | A JSON object with keys for all supported actions. No schema is imposed on the values.
+
+Existing items are updated with new values when a fetch or action produces them, with some exceptions:
+
+* Automatic fields cannot be changed.
+* If a field's previous value is non-empty and the new value is empty, the old value is kept.
+
+### Sources
+
+A source is identified by its name. A minimally functional source requires a `fetch` action that returns items.
+
+### Action API
+
+The Intake action API defines how programs should behave to be used with Intake sources.
+
+To execute an action, Intake executes the command specified by that action's `argv`.
+The process's environment is as follows:
+
+* `intake`'s environment is inherited.
+* `STATE_PATH` is set to the absolute path of a file containing the source's persistent state.
+
+When an action receives an item as input, that item's JSON representation is written to that action's `stdin`.
+When an action outputs an item, it should write the item's JSON representation to `stdout` on one line.
+All input and output is assumed to be UTF-8.
+If an item cannot be parsed or the exit code of the process is nonzero, Intake will consider the action to be a failure.
+No items will be created or updated as a result of the failed action.
+Anything written to `stderr` by the action will be captured and logged by Intake.
+
+The `fetch` action receives no input and outputs multiple items.
+This action is executed when a source is updated.
+The `fetch` action is the core of an Intake source.
+
+All other actions take an item as input and should output the same item with any modifications made by the action.
+Actions can only be executed for an item if that item has a key with the same name in its `action` field.
+The value of that key may be any non-null JSON value used to pass state to the action.
+
+The special action `on_create` is always run when an item is first returned by a fetch.
+The item does not need to declare support for `on_create`.
+This action is not accessible through the web interface, so if you need to retry the action, you should create another action with the same command as `on_create`.
+If an item's `on_create` fails, the item is still created, but without any changes made by action.
+
+The special action `on_delete` is like `on_create`, except it runs right before an item is deleted.
+It does not require explicit support and is not accessible in the web interface.
+The output of `on_delete` is ignored; it is primarily for causing side effects like managing state.

cmd/action.go

@@ -0,0 +1,34 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var actionCmd = &cobra.Command{
+	Use:   "action",
+	Short: "Manage and run source actions",
+	Long: `Add, edit, delete, and run source actions on items.
+
+A feed source is updated by the "fetch" action, which receives no input and
+returns one JSON item per line on stdout. Other source actions are run on a
+specific item, receiving that item on stdin and expecting that item, with any
+modifications made by the action, on stdout.
+
+Items declare support for an action by having an "action" key containing an
+object with a key for every supported action. The value of that key may be
+any arbitrary JSON value. Use --force to execute an unsupported action anyway,
+though the action may fail if it operates on the item's action data.
+
+The special action "on_create" is always run when an item is first returned
+by a fetch. The item does not need to declare support for "on_create". This
+action is not accessible through the web interface, so if you need to retry
+the action, you need another action with the same command as "on_create".
+If an item's "on_create" fails, the item is still created, but without any
+changes from the "on_create", if any.
+
+To execute the "fetch" action, use "intake source fetch".`,
+}
+
+func init() {
+	rootCmd.AddCommand(actionCmd)
+}

cmd/actionAdd.go

@@ -0,0 +1,52 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var actionAddCmd = &cobra.Command{
+	Use:   "add [flags] -- argv...",
+	Short: "Add an action to a source",
+	Long: `Add an action to a source.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		actionAdd(getArgv(cmd, args))
+	},
+}
+
+var actionAddSource string
+var actionAddAction string
+
+func init() {
+	actionCmd.AddCommand(actionAddCmd)
+
+	actionAddCmd.Flags().StringVarP(&actionAddSource, "source", "s", "", "Source to add action")
+	actionAddCmd.MarkFlagRequired("source")
+
+	actionAddCmd.Flags().StringVarP(&actionAddAction, "action", "a", "", "Action name")
+	actionAddCmd.MarkFlagRequired("action")
+}
+
+func actionAdd(argv []string) {
+	if actionAddSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+	if actionAddAction == "" {
+		log.Fatal("error: --action is empty")
+	}
+	if len(argv) == 0 {
+		log.Fatal("error: no argv provided")
+	}
+
+	db := openAndMigrateDb()
+
+	err := core.AddAction(db, actionAddSource, actionAddAction, argv)
+	if err != nil {
+		log.Fatalf("error: failed to add action: %v", err)
+	}
+
+	log.Printf("Added action %s to source %s", actionAddAction, actionAddSource)
+}

cmd/actionDelete.go

@@ -0,0 +1,50 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var actionDeleteCmd = &cobra.Command{
+	Use:     "delete",
+	Aliases: []string{"rm"},
+	Short:   "Delete an action from a source",
+	Long: `Delete an action from a source.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		actionDelete()
+	},
+}
+
+var actionDeleteSource string
+var actionDeleteAction string
+
+func init() {
+	actionCmd.AddCommand(actionDeleteCmd)
+
+	actionDeleteCmd.Flags().StringVarP(&actionDeleteSource, "source", "s", "", "Source to add action")
+	actionDeleteCmd.MarkFlagRequired("source")
+
+	actionDeleteCmd.Flags().StringVarP(&actionDeleteAction, "action", "a", "", "Action name")
+	actionDeleteCmd.MarkFlagRequired("action")
+}
+
+func actionDelete() {
+	if actionDeleteSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+	if actionDeleteAction == "" {
+		log.Fatal("error: --action is empty")
+	}
+
+	db := openAndMigrateDb()
+
+	err := core.DeleteAction(db, actionDeleteSource, actionDeleteAction)
+	if err != nil {
+		log.Fatalf("error: failed to delete action: %v", err)
+	}
+
+	log.Printf("Deleted action %s from source %s", actionDeleteAction, actionDeleteSource)
+}

cmd/actionEdit.go

@@ -0,0 +1,52 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var actionEditCmd = &cobra.Command{
+	Use:   "edit",
+	Short: "Edit an action on a source",
+	Long: `Edit an action on a source.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		actionEdit(getArgv(cmd, args))
+	},
+}
+
+var actionEditSource string
+var actionEditAction string
+
+func init() {
+	actionCmd.AddCommand(actionEditCmd)
+
+	actionEditCmd.Flags().StringVarP(&actionEditSource, "source", "s", "", "Source to edit action")
+	actionEditCmd.MarkFlagRequired("source")
+
+	actionEditCmd.Flags().StringVarP(&actionEditAction, "action", "a", "", "Action name")
+	actionEditCmd.MarkFlagRequired("action")
+}
+
+func actionEdit(argv []string) {
+	if actionEditSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+	if actionEditAction == "" {
+		log.Fatal("error: --action is empty")
+	}
+	if len(argv) == 0 {
+		log.Fatal("error: no argv provided")
+	}
+
+	db := openAndMigrateDb()
+
+	err := core.UpdateAction(db, actionEditSource, actionEditAction, argv)
+	if err != nil {
+		log.Fatalf("error: failed to update action: %v", err)
+	}
+
+	log.Printf("Updated action %s on source %s", actionEditAction, actionEditSource)
+}

cmd/actionExecute.go

@@ -0,0 +1,138 @@
+package cmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var actionExecuteCmd = &cobra.Command{
+	Use:     "execute",
+	Aliases: []string{"exec"},
+	Short:   "Run a source action for an item",
+	Long: fmt.Sprintf(`Execute a source action for an item.
+
+The item must declare support for the action by having the action's name
+in its "action" field. Use --force to execute the action anyway.
+
+The "fetch" action is special and does not execute for any specific item.
+Use "intake source fetch" to run the fetch action.
+
+In a dry run, the item will be printed in the chosen format and not updated.
+
+%s`, makeFormatHelpText()),
+	Run: func(cmd *cobra.Command, args []string) {
+		actionExecute()
+	},
+}
+
+var actionExecuteSource string
+var actionExecuteAction string
+var actionExecuteItem string
+var actionExecuteFormat string
+var actionExecuteDryRun bool
+var actionExecuteDiff bool
+var actionExecuteForce bool
+
+func init() {
+	actionCmd.AddCommand(actionExecuteCmd)
+
+	actionExecuteCmd.PersistentFlags().StringVarP(&actionExecuteSource, "source", "s", "", "Source of the item")
+	actionExecuteCmd.MarkFlagRequired("source")
+
+	actionExecuteCmd.PersistentFlags().StringVarP(&actionExecuteItem, "item", "i", "", "Item to run action on")
+	actionExecuteCmd.MarkFlagRequired("item")
+
+	actionExecuteCmd.PersistentFlags().StringVarP(&actionExecuteAction, "action", "a", "", "Action to run")
+	actionExecuteCmd.MarkFlagRequired("action")
+
+	actionExecuteCmd.Flags().StringVarP(&actionExecuteFormat, "format", "f", "headlines", "Feed format for returned items")
+	actionExecuteCmd.Flags().BoolVar(&actionExecuteDryRun, "dry-run", false, "Instead of updating the item, print it")
+
+	actionExecuteCmd.Flags().BoolVar(&actionExecuteDiff, "diff", false, "Show which fields of the item changed")
+
+	actionExecuteCmd.Flags().BoolVar(&actionExecuteForce, "force", false, "Execute the action even if the item does not support it")
+}
+
+func actionExecute() {
+	formatter := formatAs(actionExecuteFormat)
+
+	if actionExecuteSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+	if actionExecuteAction == "" {
+		log.Fatal("error: --action is empty")
+	}
+	if actionExecuteItem == "" {
+		log.Fatal("error: --item is empty")
+	}
+
+	db := openAndMigrateDb()
+
+	item, err := core.GetItem(db, actionExecuteSource, actionExecuteItem)
+	if err != nil {
+		log.Fatalf("error: failed to get item: %v", err)
+	}
+
+	if item.Action[actionExecuteAction] == nil {
+		if actionExecuteForce {
+			log.Printf("warning: force-executing %s on %s/%s", actionExecuteAction, actionExecuteSource, actionExecuteItem)
+		} else {
+			log.Fatalf("error: %s/%s does not support %s", actionExecuteSource, actionExecuteItem, actionExecuteAction)
+		}
+	}
+
+	argv, err := core.GetArgvForAction(db, actionExecuteSource, actionExecuteAction)
+	if err != nil {
+		log.Fatalf("error: failed to get action: %v", err)
+	}
+
+	itemJson, err := json.Marshal(item)
+	if err != nil {
+		log.Fatalf("error: failed to serialize item: %v", err)
+	}
+
+	res, err := core.Execute(actionExecuteSource, argv, nil, string(itemJson), time.Minute)
+	if err != nil {
+		log.Fatalf("error: failed to execute action: %v", err)
+	}
+	if len(res) != 1 {
+		log.Fatalf("error: expected action to produce exactly one item, got %d", len(res))
+	}
+	newItem := res[0]
+	core.BackfillItem(&newItem, &item)
+
+	if actionExecuteDiff {
+		if item.Title != newItem.Title {
+			log.Printf("title: %s => %s", item.Title, newItem.Title)
+		}
+		if item.Author != newItem.Author {
+			log.Printf("author: %s => %s", item.Author, newItem.Author)
+		}
+		if item.Body != newItem.Body {
+			log.Printf("body: %s => %s", item.Body, newItem.Body)
+		}
+		if item.Link != newItem.Link {
+			log.Printf("link: %s => %s", item.Link, newItem.Link)
+		}
+		if item.Time != newItem.Time {
+			log.Printf("time: %d => %d", item.Time, newItem.Time)
+		}
+		if core.ItemsAreEqual(item, newItem) {
+			log.Printf("no changes\n")
+		}
+	}
+
+	if actionExecuteDryRun {
+		fmt.Println(formatter(res[0]))
+		return
+	}
+
+	if err = core.UpdateItems(db, []core.Item{newItem}); err != nil {
+		log.Fatalf("error: failed to update item: %v", err)
+	}
+}

cmd/actionList.go

@@ -0,0 +1,66 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+	"slices"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var actionListCmd = &cobra.Command{
+	Use:     "list",
+	Aliases: []string{"ls"},
+	Short:   "List actions on a source",
+	Long: `List actions on a source.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		actionList()
+	},
+}
+
+var actionListSource string
+var actionListArgv bool
+
+func init() {
+	actionCmd.AddCommand(actionListCmd)
+
+	actionListCmd.Flags().StringVarP(&actionListSource, "source", "s", "", "Source to list actions")
+	actionListCmd.MarkFlagRequired("source")
+
+	actionListCmd.Flags().BoolVarP(&actionListArgv, "argv", "a", false, "Include action command")
+}
+
+func actionList() {
+	if actionListSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+
+	db := openAndMigrateDb()
+
+	actions, err := core.GetActionsForSource(db, actionListSource)
+	if err != nil {
+		log.Fatal(err)
+	}
+	slices.SortFunc(actions, actionSort)
+
+	if actionListArgv {
+		actionArgv := make(map[string][]string)
+		for _, name := range actions {
+			argv, err := core.GetArgvForAction(db, actionListSource, name)
+			if err != nil {
+				log.Fatalf("error: could not get argv for source %s action %s: %v", actionListSource, name, err)
+			}
+			actionArgv[name] = argv
+		}
+		for _, name := range actions {
+			fmt.Printf("%s %v\n", name, actionArgv[name])
+		}
+
+	} else {
+		for _, action := range actions {
+			fmt.Println(action)
+		}
+	}
+}

cmd/channel.go

@@ -0,0 +1,16 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var channelCmd = &cobra.Command{
+	Use:   "channel",
+	Short: "Manage channels",
+	Long: `
+`,
+}
+
+func init() {
+	rootCmd.AddCommand(channelCmd)
+}

cmd/channelAdd.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var channelAddCmd = &cobra.Command{
+	Use:   "add",
+	Short: "Create a channel",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	channelCmd.AddCommand(channelAddCmd)
+}

cmd/channelDelete.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var channelDeleteCmd = &cobra.Command{
+	Use:   "delete",
+	Short: "Delete a channel",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	channelCmd.AddCommand(channelDeleteCmd)
+}

cmd/channelEdit.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var channelEditCmd = &cobra.Command{
+	Use:   "edit",
+	Short: "Edit a channel",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	channelCmd.AddCommand(channelEditCmd)
+}

cmd/feed.go

@@ -0,0 +1,70 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var feedCmd = &cobra.Command{
+	Use:   "feed",
+	Short: "Display the item feed",
+	Long: fmt.Sprintf(`Display the intake item feed in various formats.
+The default format is "headlines".
+
+%s`, makeFormatHelpText()),
+	Run: func(cmd *cobra.Command, args []string) {
+		feed()
+	},
+}
+
+var feedFormat string
+var feedSource string
+var feedChannel string
+var feedShowInactive bool
+
+func init() {
+	rootCmd.AddCommand(feedCmd)
+
+	feedCmd.Flags().StringVarP(&feedFormat, "format", "f", "headlines", "Feed format")
+	feedCmd.Flags().StringVarP(&feedSource, "source", "s", "", "Limit to items from source")
+	feedCmd.Flags().StringVarP(&feedChannel, "channel", "c", "", "Limit to items from channel")
+	feedCmd.MarkFlagsMutuallyExclusive("source", "channel")
+	feedCmd.Flags().BoolVar(&feedShowInactive, "all", false, "Show inactive items")
+}
+
+func feed() {
+	formatter := formatAs(feedFormat)
+
+	db := openAndMigrateDb()
+
+	var items []core.Item
+	var err error
+	if feedSource != "" {
+		if feedShowInactive {
+			items, err = core.GetAllItemsForSource(db, feedSource)
+		} else {
+			items, err = core.GetActiveItemsForSource(db, feedSource)
+		}
+		if err != nil {
+			log.Fatalf("error: failed to fetch items from %s:, %v", feedSource, err)
+		}
+	} else if feedChannel != "" {
+		log.Fatal("error: unimplemented")
+	} else {
+		if feedShowInactive {
+			items, err = core.GetAllItems(db)
+		} else {
+			items, err = core.GetAllActiveItems(db)
+		}
+		if err != nil {
+			log.Fatalf("error: failed to fetch items: %v", err)
+		}
+	}
+
+	for _, item := range items {
+		fmt.Println(formatter(item))
+	}
+}

cmd/item.go

@@ -0,0 +1,16 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var itemCmd = &cobra.Command{
+	Use:   "item",
+	Short: "Manage items",
+	Long: `Add, edit, or deactivate items.
+`,
+}
+
+func init() {
+	rootCmd.AddCommand(itemCmd)
+}

cmd/itemAdd.go

@@ -0,0 +1,85 @@
+package cmd
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/spf13/cobra"
+)
+
+var itemAddCmd = &cobra.Command{
+	Use:   "add",
+	Short: "Add an item",
+	Long: `Create an ad-hoc item in a source.
+
+By default, the item is created in the "default" source, which is created
+if it doesn't exist, with a random id.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		itemAdd()
+	},
+}
+
+var addItemSource string
+var addItemId string
+var addItemTitle string
+var addItemAuthor string
+var addItemBody string
+var addItemLink string
+var addItemTime int
+var addItemActions string
+
+func init() {
+	itemCmd.AddCommand(itemAddCmd)
+
+	itemAddCmd.Flags().StringVarP(&addItemSource, "source", "s", "", "Source in which to create the item (default: default)")
+	itemAddCmd.Flags().StringVarP(&addItemId, "id", "i", "", "Item id (default: random hex)")
+	itemAddCmd.Flags().StringVarP(&addItemTitle, "title", "t", "", "Item title")
+	itemAddCmd.Flags().StringVarP(&addItemAuthor, "author", "a", "", "Item author")
+	itemAddCmd.Flags().StringVarP(&addItemBody, "body", "b", "", "Item body")
+	itemAddCmd.Flags().StringVarP(&addItemLink, "link", "l", "", "Item link")
+	itemAddCmd.Flags().IntVarP(&addItemTime, "time", "m", 0, "Item time as a Unix timestamp")
+	itemAddCmd.Flags().StringVarP(&addItemActions, "action", "x", "", "Item time as a Unix timestamp")
+}
+
+func itemAdd() {
+	// Default to "default" source
+	if addItemSource == "" {
+		addItemSource = "default"
+	}
+	// Default id to random hex string
+	if addItemId == "" {
+		bytes := make([]byte, 16)
+		if _, err := rand.Read(bytes); err != nil {
+			log.Fatalf("error: failed to generate id: %v", err)
+		}
+		addItemId = hex.EncodeToString(bytes)
+	}
+
+	var actions core.Actions
+	if addItemActions != "" {
+		if err := json.Unmarshal([]byte(addItemActions), &actions); err != nil {
+			log.Fatalf("error: could not parse actions: %v", err)
+		}
+	}
+
+	db := openAndMigrateDb()
+
+	if err := core.AddItems(db, []core.Item{{
+		Source: addItemSource,
+		Id:     addItemId,
+		Title:  addItemTitle,
+		Author: addItemAuthor,
+		Body:   addItemBody,
+		Link:   addItemLink,
+		Time:   addItemTime,
+		Action: actions,
+	}}); err != nil {
+		log.Fatalf("error: failed to add item: %s", err)
+	}
+
+	log.Printf("Added %s/%s\n", addItemSource, addItemId)
+}

cmd/itemDeactivate.go

@@ -0,0 +1,46 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/spf13/cobra"
+)
+
+var itemDeactivateCmd = &cobra.Command{
+	Use:     "deactivate",
+	Aliases: []string{"deac"},
+	Short:   "Deactivate an item",
+	Long: `Deactivate items, hiding them from feeds and marking them for deletion.
+	
+Deactivation is idempotent.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		itemDeactivate()
+	},
+}
+
+var deacSource string
+var deacItem string
+
+func init() {
+	itemCmd.AddCommand(itemDeactivateCmd)
+
+	itemDeactivateCmd.Flags().StringVarP(&deacSource, "source", "s", "", "Source of the item")
+	itemDeactivateCmd.MarkFlagRequired("source")
+	itemDeactivateCmd.Flags().StringVarP(&deacItem, "item", "i", "", "Item id")
+	itemDeactivateCmd.MarkFlagRequired("item")
+}
+
+func itemDeactivate() {
+	db := openAndMigrateDb()
+
+	active, err := core.DeactivateItem(db, deacSource, deacItem)
+	if err != nil {
+		log.Fatalf("Failed to deactivate item: %s", err)
+	}
+	if active {
+		fmt.Printf("Deactivated %s/%s\n", deacSource, deacItem)
+	}
+}

cmd/itemEdit.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var itemEditCmd = &cobra.Command{
+	Use:   "edit",
+	Short: "Edit an item",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	itemCmd.AddCommand(itemEditCmd)
+}

cmd/migrate.go

@@ -0,0 +1,50 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/spf13/cobra"
+)
+
+var migrateCmd = &cobra.Command{
+	Use:   "migrate",
+	Short: "Migrate an intake database to the latest version",
+	Long: `Migrate an intake database to the latest version.
+	
+Note that the database will be created if it does not exist, even with --list.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		migrate()
+	},
+}
+
+var migrateListOnly bool
+
+func init() {
+	rootCmd.AddCommand(migrateCmd)
+
+	migrateCmd.Flags().BoolVarP(&migrateListOnly, "list", "l", false, "Show the list of migrations")
+}
+
+func migrate() {
+	db := openDb()
+
+	core.InitDatabase(db)
+	if migrateListOnly {
+		pending, err := core.GetPendingMigrations(db)
+		if err != nil {
+			log.Fatal(err)
+		}
+		for name, complete := range pending {
+			if complete {
+				fmt.Printf("[x] %s\n", name)
+			} else {
+				fmt.Printf("[ ] %s\n", name)
+			}
+		}
+	} else {
+		core.MigrateDatabase(db)
+	}
+}

cmd/passwd.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var passwdCmd = &cobra.Command{
+	Use:   "passwd",
+	Short: "Set the password for the web interface",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(passwdCmd)
+}

cmd/root.go

@@ -0,0 +1,109 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"strings"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var rootCmd = &cobra.Command{
+	Use:   "intake",
+	Short: "Universal and extensible feed aggregator",
+	Long:  `intake, the universal and extensible feed aggregator`,
+}
+
+func Execute() {
+	err := rootCmd.Execute()
+	if err != nil {
+		os.Exit(1)
+	}
+}
+
+var dataPath string
+
+func init() {
+	// Disable the automatic help command
+	rootCmd.SetHelpCommand(&cobra.Command{Hidden: true})
+
+	// All commands need to operate on a database
+	rootCmd.PersistentFlags().StringVarP(&dataPath, "data-dir", "d", "", "Path to the intake data directory containing the database")
+}
+
+//
+// Common logic shared by multiple commands
+//
+
+func getDbPath() string {
+	if dataPath != "" {
+		return core.DatabasePath(dataPath)
+	}
+	if dataDir := core.ResolveDataDir(); dataDir != "" {
+		return core.DatabasePath(dataDir)
+	}
+	fmt.Println("error: no database specified")
+	fmt.Println("One of --data-dir, INTAKE_DATA_DIR, XDG_DATA_HOME, or HOME must be defined.")
+	os.Exit(1)
+	return ""
+}
+
+// Attempt to open the specified database and exit with an error if it fails.
+func openDb() *core.DB {
+	dbPath := getDbPath()
+	db, err := core.OpenDb(dbPath)
+	if err != nil {
+		log.Fatalf("error: failed to open %s", dbPath)
+	}
+	return db
+}
+
+// Attempt to open and migrate the specified database and exit with an error if it fails.
+func openAndMigrateDb() *core.DB {
+	db := openDb()
+	if err := core.InitDatabase(db); err != nil {
+		log.Fatalf("error: failed to init database: %v", err)
+	}
+	if err := core.MigrateDatabase(db); err != nil {
+		log.Fatalf("error: failed to migrate database: %v", err)
+	}
+	return db
+}
+
+func getArgv(cmd *cobra.Command, args []string) []string {
+	lenAtDash := cmd.Flags().ArgsLenAtDash()
+	if lenAtDash == -1 {
+		return nil
+	} else {
+		return args[lenAtDash:]
+	}
+}
+
+// Sort "fetch" action ahead of other actions
+func actionSort(a string, b string) int {
+	if a == "fetch" {
+		return -1
+	}
+	if b == "fetch" {
+		return 1
+	}
+	return strings.Compare(a, b)
+}
+
+func makeFormatHelpText() string {
+	text := "Available formats:\n"
+	for format, desc := range core.AvailableFormats {
+		text += fmt.Sprintf("  %-13s %s\n", format, desc)
+	}
+	return text
+}
+
+func formatAs(format string) func(item core.Item) string {
+	formatter, err := core.FormatAs(format)
+	if err != nil {
+		log.Fatalf("error: %v", err)
+	}
+	return formatter
+}

cmd/serve.go

@@ -0,0 +1,31 @@
+package cmd
+
+import (
+	"github.com/Jaculabilis/intake/web"
+	"github.com/spf13/cobra"
+)
+
+var serveCmd = &cobra.Command{
+	Use:   "serve",
+	Short: "Serve the web interface",
+	Long: `Serve the intake web interface.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		serve()
+	},
+}
+
+var serveAddr string
+var servePort string
+
+func init() {
+	rootCmd.AddCommand(serveCmd)
+
+	serveCmd.Flags().StringVarP(&serveAddr, "addr", "a", "localhost", "Address to bind to")
+	serveCmd.Flags().StringVarP(&servePort, "port", "p", "8081", "Port to bind to")
+}
+
+func serve() {
+	db := openAndMigrateDb()
+	web.RunServer(db, serveAddr, servePort)
+}

cmd/source.go

@@ -0,0 +1,23 @@
+/*
+Copyright © 2025 NAME HERE <EMAIL ADDRESS>
+*/
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var sourceCmd = &cobra.Command{
+	Use:   "source",
+	Short: "Manage sources",
+	Long: `Manage sources.
+
+A source represents a single content feed that generates discrete feed items.
+The command defined in the "fetch" action is used to check for new items to
+update the feed.
+`,
+}
+
+func init() {
+	rootCmd.AddCommand(sourceCmd)
+}

cmd/sourceAdd.go

@@ -0,0 +1,41 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var sourceAddCmd = &cobra.Command{
+	Use:   "add",
+	Short: "Create a source",
+	Long: `Create a source.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		sourceAdd()
+	},
+}
+
+var sourceAddSource string
+
+func init() {
+	sourceCmd.AddCommand(sourceAddCmd)
+
+	sourceAddCmd.Flags().StringVarP(&sourceAddSource, "source", "s", "", "Source name")
+	sourceAddCmd.MarkFlagRequired("source")
+}
+
+func sourceAdd() {
+	if sourceAddSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+
+	db := openAndMigrateDb()
+
+	if err := core.AddSource(db, sourceAddSource); err != nil {
+		log.Fatalf("error: failed to add source: %v", err)
+	}
+
+	log.Printf("Added source %s", sourceAddSource)
+}

cmd/sourceDeactivate.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var sourceDeactivateCmd = &cobra.Command{
+	Use:   "deactivate",
+	Short: "Deactivate all items in a source",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	sourceCmd.AddCommand(sourceDeactivateCmd)
+}

cmd/sourceDelete.go

@@ -0,0 +1,41 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var sourceDeleteCmd = &cobra.Command{
+	Use:     "delete",
+	Aliases: []string{"rm"},
+	Short:   "Delete a source",
+	Long: `Delete a source.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		sourceDelete()
+	},
+}
+
+var sourceDeleteSource string
+
+func init() {
+	sourceCmd.AddCommand(sourceDeleteCmd)
+
+	sourceDeleteCmd.Flags().StringVarP(&sourceDeleteSource, "source", "s", "", "Source to delete")
+}
+
+func sourceDelete() {
+	if sourceDeleteSource == "" {
+		log.Fatal("error: --source is empty")
+	}
+
+	db := openAndMigrateDb()
+
+	if err := core.DeleteSource(db, sourceDeleteSource); err != nil {
+		log.Fatalf("error: failed to delete source: %v", err)
+	}
+
+	log.Printf("Deleted source %s", sourceDeleteSource)
+}

cmd/sourceEdit.go

@@ -0,0 +1,21 @@
+package cmd
+
+import (
+	"log"
+
+	"github.com/spf13/cobra"
+)
+
+var sourceEditCmd = &cobra.Command{
+	Use:   "edit",
+	Short: "Edit a source",
+	Long: `
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Fatal("not implemented")
+	},
+}
+
+func init() {
+	sourceCmd.AddCommand(sourceEditCmd)
+}

cmd/sourceFetch.go

@@ -0,0 +1,72 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var sourceFetchCmd = &cobra.Command{
+	Use:   "fetch",
+	Short: "Fetch items for a source and update the feed",
+	Long: fmt.Sprintf(`Fetch items from a feed source using the configured "fetch" action.
+Items returned by a successful fetch will be used to update the source.
+A fetch is successful if all items output by the fetch are parsed successfully
+and the exit code is 0. No changes will be made to the source if the fetch
+does not succeed.
+
+In a dry run, the items will be printed according to the chosen format and
+the source will not be updated with the fetch result.
+
+%s`, makeFormatHelpText()),
+	Run: func(cmd *cobra.Command, args []string) {
+		sourceFetch()
+	},
+}
+
+var sourceFetchSource string
+var sourceFetchFormat string
+var sourceFetchDryRun bool
+
+func init() {
+	sourceCmd.AddCommand(sourceFetchCmd)
+
+	sourceFetchCmd.Flags().StringVarP(&sourceFetchSource, "source", "s", "", "Source name to fetch (required)")
+	sourceFetchCmd.MarkFlagRequired("source")
+
+	sourceFetchCmd.Flags().StringVarP(&sourceFetchFormat, "format", "f", "headlines", "Feed format for returned items.")
+	sourceFetchCmd.Flags().BoolVar(&sourceFetchDryRun, "dry-run", false, "Instead of updating the source, print the fetched items")
+}
+
+func sourceFetch() {
+	formatter := formatAs(sourceFetchFormat)
+
+	db := openAndMigrateDb()
+
+	argv, err := core.GetArgvForAction(db, sourceFetchSource, "fetch")
+	if err != nil {
+		log.Fatalf("error: failed to get fetch action: %v", err)
+	}
+
+	items, err := core.Execute(sourceFetchSource, argv, nil, "", time.Minute)
+	if err != nil {
+		log.Fatalf("error: failed to execute fetch: %v", err)
+	}
+
+	if sourceFetchDryRun {
+		log.Printf("Fetch returned %d items", len(items))
+		for _, item := range items {
+			fmt.Println(formatter(item))
+		}
+		return
+	}
+
+	added, deleted, err := core.UpdateWithFetchedItems(db, sourceFetchSource, items)
+	if err != nil {
+		log.Fatalf("error: failed to update: %v", err)
+	}
+	log.Printf("%s added %d items, updated %d items, and deleted %d items", sourceFetchSource, added, len(items)-added, deleted)
+}

cmd/sourceList.go

@@ -0,0 +1,58 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+	"slices"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var sourceListCmd = &cobra.Command{
+	Use:     "list",
+	Aliases: []string{"ls"},
+	Short:   "List sources",
+	Long: `Print the list of sources.
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		sourceList()
+	},
+}
+
+var sourceListShowActions bool
+
+func init() {
+	sourceCmd.AddCommand(sourceListCmd)
+
+	sourceListCmd.Flags().BoolVarP(&sourceListShowActions, "actions", "a", false, "Include source actions")
+}
+
+func sourceList() {
+	db := openAndMigrateDb()
+
+	names, err := core.GetSources(db)
+	if err != nil {
+		log.Fatalf("error: failed to get sources: %v", err)
+	}
+	slices.Sort(names)
+
+	if sourceListShowActions {
+		sourceActions := make(map[string][]string)
+		for _, name := range names {
+			actions, err := core.GetActionsForSource(db, name)
+			if err != nil {
+				log.Fatalf("error: could not get actions for source %s: %v", name, err)
+			}
+			slices.SortFunc(actions, actionSort)
+			sourceActions[name] = actions
+		}
+		for _, name := range names {
+			fmt.Printf("%s %v\n", name, sourceActions[name])
+		}
+	} else {
+		for _, name := range names {
+			fmt.Println(name)
+		}
+	}
+}

cmd/sourceTest.go

@@ -0,0 +1,49 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/spf13/cobra"
+)
+
+var sourceTestCmd = &cobra.Command{
+	Use:   "test [flags] -- argv",
+	Short: "Test a fetch action",
+	Long: fmt.Sprintf(`Execute a command as if it were a feed source's fetch action.
+
+%s`, makeFormatHelpText()),
+	Run: func(cmd *cobra.Command, args []string) {
+		l := cmd.Flags().ArgsLenAtDash()
+		if l == -1 {
+			sourceTest(nil)
+		} else {
+			sourceTest(args[l:])
+		}
+	},
+}
+
+var sourceTestEnv []string
+var sourceTestFormat string
+
+func init() {
+	sourceCmd.AddCommand(sourceTestCmd)
+
+	sourceTestCmd.Flags().StringArrayVarP(&sourceTestEnv, "env", "e", nil, "Environment variables to set, in the form KEY=VAL")
+	sourceTestCmd.Flags().StringVarP(&sourceTestFormat, "format", "f", "headlines", "Feed format for returned items.")
+}
+
+func sourceTest(cmd []string) {
+	formatter := formatAs(sourceTestFormat)
+
+	items, err := core.Execute("", cmd, sourceTestEnv, "", time.Minute)
+	log.Printf("Returned %d items", len(items))
+	if err != nil {
+		log.Fatal(err)
+	}
+	for _, item := range items {
+		fmt.Println(formatter(item))
+	}
+}

core/action.go

@@ -0,0 +1,208 @@
+package core
+
+import (
+	"bufio"
+	"context"
+	"database/sql/driver"
+	"encoding/json"
+	"errors"
+	"io"
+	"log"
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+)
+
+// Type alias for storing string array as jsonb
+type argList []string
+
+func (a argList) Value() (driver.Value, error) {
+	return json.Marshal(a)
+}
+
+func (a *argList) Scan(value interface{}) error {
+	return json.Unmarshal([]byte(value.(string)), a)
+}
+
+func AddAction(db *DB, source string, name string, argv []string) error {
+	_, err := db.Exec(`
+		insert into actions (source, name, argv)
+		values (?, ?, jsonb(?))
+	`, source, name, argList(argv))
+	return err
+}
+
+func UpdateAction(db *DB, source string, name string, argv []string) error {
+	_, err := db.Exec(`
+		update actions
+		set argv = jsonb(?)
+		where source = ? and name = ?
+	`, argList(argv), source, name)
+	return err
+}
+
+func GetActionsForSource(db *DB, source string) ([]string, error) {
+	rows, err := db.Query(`
+		select name
+		from actions
+		where source = ?
+	`, source)
+	if err != nil {
+		return nil, err
+	}
+	var names []string
+	for rows.Next() {
+		var name string
+		err = rows.Scan(&name)
+		if err != nil {
+			return nil, err
+		}
+		names = append(names, name)
+	}
+	return names, nil
+}
+
+func GetArgvForAction(db *DB, source string, name string) ([]string, error) {
+	rows := db.QueryRow(`
+		select json(argv)
+		from actions
+		where source = ? and name = ?
+	`, source, name)
+	var argv argList
+	err := rows.Scan(&argv)
+	if err != nil {
+		return nil, err
+	}
+	return argv, nil
+}
+
+func DeleteAction(db *DB, source string, name string) error {
+	_, err := db.Exec(`
+		delete from actions
+		where source = ? and name = ?
+	`, 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,
+	input string,
+	timeout time.Duration,
+) ([]Item, error) {
+	log.Printf("Executing %v", argv)
+
+	if len(argv) == 0 {
+		return nil, errors.New("empty argv")
+	}
+	if source == "" {
+		return nil, errors.New("empty source")
+	}
+
+	env = append(env, "STATE_PATH=")
+
+	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, err
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return 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, 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, 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, err
+	} else if err != nil {
+		log.Printf("error: %s failed with error: %s\n", argv[0], err)
+		return nil, err
+	}
+
+	if parseError {
+		log.Printf("error: could not parse item\n")
+		return nil, errors.New("invalid JSON")
+	}
+
+	return items, nil
+}

core/action_test.go

@@ -0,0 +1,182 @@
+package core
+
+import (
+	"testing"
+	"time"
+)
+
+func TestActionCreate(t *testing.T) {
+	db := EphemeralDb(t)
+
+	if err := AddAction(db, "test", "hello", []string{"echo", "hello"}); err == nil {
+		t.Fatal("Action created for nonexistent source")
+	}
+
+	if err := AddSource(db, "test"); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := AddAction(db, "test", "hello", []string{"echo", "hello"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := AddAction(db, "test", "goodbye", []string{"exit", "1"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := UpdateAction(db, "test", "goodbye", []string{"echo", "goodbye"}); err != nil {
+		t.Fatal(err)
+	}
+
+	actions, err := GetActionsForSource(db, "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(actions) != 2 {
+		t.Fatal("expected 2 actions")
+	}
+	found := make(map[string]bool)
+	for _, action := range actions {
+		found[action] = true
+	}
+	if !found["hello"] || !found["goodbye"] {
+		t.Fatalf("missing hello and/or goodbye, got: %v", actions)
+	}
+
+	argv, err := GetArgvForAction(db, "test", "goodbye")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(argv) != 2 || argv[0] != "echo" || argv[1] != "goodbye" {
+		t.Fatalf("expected [echo goodbye], got: %v", argv)
+	}
+
+	err = DeleteAction(db, "test", "hello")
+	if err != nil {
+		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) {
+		return Execute("_", argv, nil, "", time.Minute)
+	}
+
+	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, "", 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, "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"}, "", 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")
+	}
+}

core/data.go

@@ -0,0 +1,21 @@
+package core
+
+import (
+	"os"
+	"path/filepath"
+)
+
+func ResolveDataDir() string {
+	if intakeData := os.Getenv("INTAKE_DATA_DIR"); intakeData != "" {
+		return intakeData
+	} else if xdgData := os.Getenv("XDG_DATA_HOME"); xdgData != "" {
+		return filepath.Join(xdgData, "intake")
+	} else if home := os.Getenv("HOME"); home != "" {
+		return filepath.Join(home, ".local", "share", "intake")
+	}
+	return ""
+}
+
+func DatabasePath(dataDir string) string {
+	return filepath.Join(dataDir, "intake.db")
+}

core/db.go

@@ -0,0 +1,89 @@
+package core
+
+import (
+	"database/sql"
+	"runtime"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+type DB struct {
+	ro *sql.DB
+	rw *sql.DB
+}
+
+func (db *DB) Query(query string, args ...any) (*sql.Rows, error) {
+	return db.ro.Query(query, args...)
+}
+
+func (db *DB) QueryRow(query string, args ...any) *sql.Row {
+	return db.ro.QueryRow(query, args...)
+}
+
+func (db *DB) Exec(query string, args ...any) (sql.Result, error) {
+	return db.rw.Exec(query, args...)
+}
+
+func (db *DB) Transact(transaction func(*sql.Tx) error) error {
+	tx, err := db.rw.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	_, err = tx.Exec("rollback; begin immediate")
+	if err != nil {
+		return err
+	}
+	if err = transaction(tx); err != nil {
+		return err
+	}
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func defaultPragma(db *sql.DB) (sql.Result, error) {
+	return db.Exec(`
+		pragma journal_mode = WAL;
+		pragma busy_timeout = 5000;
+		pragma synchronous = NORMAL;
+		pragma cache_size = 1000000000;
+		pragma foreign_keys = true;
+		pragma temp_store = memory;
+		pragma mmap_size = 3000000000;
+	`)
+}
+
+func OpenDb(dataSourceName string) (*DB, error) {
+	ro, err := sql.Open("sqlite3", dataSourceName)
+	if err != nil {
+		defer ro.Close()
+		return nil, err
+	}
+	ro.SetMaxOpenConns(max(4, runtime.NumCPU()))
+	_, err = defaultPragma(ro)
+	if err != nil {
+		defer ro.Close()
+		return nil, err
+	}
+
+	rw, err := sql.Open("sqlite3", dataSourceName)
+	if err != nil {
+		defer ro.Close()
+		defer rw.Close()
+		return nil, err
+	}
+	rw.SetMaxOpenConns(1)
+	_, err = defaultPragma(rw)
+	if err != nil {
+		defer ro.Close()
+		defer rw.Close()
+		return nil, err
+	}
+
+	wrapper := new(DB)
+	wrapper.ro = ro
+	wrapper.rw = rw
+	return wrapper, nil
+}

core/db_test.go

@@ -0,0 +1,116 @@
+package core
+
+import (
+	"database/sql"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func TestDeleteSourceCascade(t *testing.T) {
+	db := EphemeralDb(t)
+
+	if err := AddSource(db, "source1"); err != nil {
+		t.Fatalf("failed to add source1: %v", err)
+	}
+	if err := AddSource(db, "source2"); err != nil {
+		t.Fatalf("failed to add source2: %v", err)
+	}
+	if err := AddItems(db, []Item{
+		{"source1", "item1", 0, true, "", "", "", "", 0, nil},
+		{"source2", "item2", 0, true, "", "", "", "", 0, nil},
+	}); err != nil {
+		t.Fatalf("failed to add items: %v", err)
+	}
+
+	items, err := GetAllActiveItems(db)
+	if err != nil {
+		t.Fatalf("failed to get active items: %v", err)
+	}
+	if len(items) != 2 {
+		t.Fatal("Expected 2 items")
+	}
+
+	if err := DeleteSource(db, "source1"); err != nil {
+		t.Fatal(err)
+	}
+	items, err = GetAllActiveItems(db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(items) != 1 {
+		t.Fatalf("Expected only 1 item after source delete, got %d", len(items))
+	}
+
+	err = AddItems(db, []Item{{"source1", "item3", 0, true, "", "", "", "", 0, nil}})
+	if err == nil {
+		t.Fatal("Unexpected success adding item for nonexistent source")
+	}
+}
+
+func TestTransaction(t *testing.T) {
+	db := EphemeralDb(t)
+	if _, err := db.Exec("create table planets (name text) strict"); err != nil {
+		t.Fatal(err)
+	}
+
+	// A transaction that should succeed
+	err := db.Transact(func(tx *sql.Tx) error {
+		if _, err := tx.Exec("insert into planets (name) values (?)", "mercury"); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := tx.Exec("insert into planets (name) values (?)", "venus"); err != nil {
+			t.Fatal(err)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Check both rows were inserted
+	rows, err := db.Query("select name from planets")
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := map[string]bool{}
+	for rows.Next() {
+		var name string
+		if err = rows.Scan(&name); err != nil {
+			t.Fatal(err)
+		}
+		found[name] = true
+	}
+	if !found["mercury"] || !found["venus"] {
+		t.Fatal("transaction failed to insert rows")
+	}
+
+	// A transaction that should fail
+	err = db.Transact(func(tx *sql.Tx) error {
+		if _, err := tx.Exec("insert into planets (name) values (?)", "earth"); err != nil {
+			t.Fatal(err)
+		}
+		_, err := tx.Exec("insert into planets (name) values (?, ?)", "moon", "surprise asteroid!")
+		return err
+	})
+	if err == nil {
+		t.Fatal("expected error")
+	}
+
+	// Check the third insert was rolled back by the error
+	rows, err = db.Query("select name from planets")
+	if err != nil {
+		t.Fatal(err)
+	}
+	found = map[string]bool{}
+	for rows.Next() {
+		var name string
+		if err = rows.Scan(&name); err != nil {
+			t.Fatal(err)
+		}
+		found[name] = true
+	}
+	if found["earth"] {
+		t.Fatal("transaction failed to roll back insert")
+	}
+}

core/item.go

@@ -0,0 +1,80 @@
+package core
+
+import (
+	"database/sql/driver"
+	"encoding/json"
+	"fmt"
+	"log"
+)
+
+type Actions map[string]json.RawMessage
+
+func (a Actions) Value() (driver.Value, error) {
+	return json.Marshal(a)
+}
+
+func (a *Actions) Scan(value interface{}) error {
+	return json.Unmarshal([]byte(value.(string)), a)
+}
+
+type Item struct {
+	Source  string  `json:"source"`
+	Id      string  `json:"id"`
+	Created int     `json:"created"`
+	Active  bool    `json:"active"`
+	Title   string  `json:"title"`
+	Author  string  `json:"author"`
+	Body    string  `json:"body"`
+	Link    string  `json:"link"`
+	Time    int     `json:"time"`
+	Action  Actions `json:"action"`
+}
+
+// Whether an item that no longer appears in a fetch can be deleted.
+func (item Item) Deletable() bool {
+	return !item.Active
+}
+
+func ItemsAreEqual(first Item, second Item) bool {
+	// Hacky but easy to use
+	return fmt.Sprintf("%#v", first) == fmt.Sprintf("%#v", second)
+}
+
+func FormatAsHeadline(item Item) string {
+	title := item.Title
+	if title == "" {
+		title = item.Id
+	}
+	return title
+}
+
+func FormatAsJson(item Item) string {
+	data, err := json.Marshal(item)
+	if err != nil {
+		log.Fatalf("error: failed to serialize %s/%s: %v", item.Source, item.Id, err)
+	}
+	return string(data)
+}
+
+func FormatAsShort(item Item) string {
+	return fmt.Sprintf("%s/%s", item.Source, item.Id)
+}
+
+func FormatAs(format string) (func(item Item) string, error) {
+	switch format {
+	case "headlines":
+		return FormatAsHeadline, nil
+	case "json":
+		return FormatAsJson, nil
+	case "short":
+		return FormatAsShort, nil
+	default:
+		return nil, fmt.Errorf("invalid format '%s'", format)
+	}
+}
+
+var AvailableFormats = map[string]string{
+	"headlines": "Only item titles",
+	"json":      "Full item JSON",
+	"short":     "Item source and id",
+}

core/item_test.go

@@ -0,0 +1,51 @@
+package core
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestItemFormatsExist(t *testing.T) {
+	for name := range AvailableFormats {
+		formatter, err := FormatAs(name)
+		if err != nil {
+			t.Fatalf("error getting formatter for available format %s: %v", name, err)
+		}
+		if formatter == nil {
+			t.Fatalf("formatter %s is nil", name)
+		}
+	}
+}
+
+func TestItemRoundTrip(t *testing.T) {
+	db := EphemeralDb(t)
+	if err := AddSource(db, "_"); err != nil {
+		t.Fatalf("failed to create source: %v", err)
+	}
+
+	item1 := Item{
+		Source:  "_",
+		Id:      "a",
+		Created: 0,
+		Active:  true,
+		Title:   "title",
+		Author:  "author",
+		Body:    "body",
+		Link:    "link",
+		Time:    123456,
+		Action: map[string]json.RawMessage{
+			"hello": json.RawMessage(`"world"`),
+		},
+	}
+	if err := AddItems(db, []Item{item1}); err != nil {
+		t.Fatalf("update failed: %v", err)
+	}
+	item2, err := GetItem(db, item1.Source, item1.Id)
+	if err != nil {
+		t.Fatalf("could not get item: %v", err)
+	}
+	item2.Created = 0 // automatically set by db
+	if !ItemsAreEqual(item1, item2) {
+		t.Fatalf("items are not equal, err %v", err)
+	}
+}

core/migrations.go

@@ -0,0 +1,101 @@
+package core
+
+import (
+	"embed"
+	"log"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+//go:embed sql/*.sql
+var migrations embed.FS
+
+// Idempotently initialize the database. Safe to call unconditionally.
+func InitDatabase(db *DB) error {
+	rows, err := db.Query(`
+		select exists (
+			select 1
+			from sqlite_master
+			where type = 'table'
+			and name = 'migrations'
+		)
+	`)
+	if err != nil {
+		return err
+	}
+
+	var exists bool
+	for rows.Next() {
+		err = rows.Scan(&exists)
+		if err != nil {
+			return err
+		}
+	}
+
+	if exists {
+		return nil
+	}
+
+	err = ApplyMigration(db, "0000_baseline.sql")
+	return err
+}
+
+// Get a map of migration names to whether the migration has been applied.
+func GetPendingMigrations(db *DB) (map[string]bool, error) {
+	allMigrations, err := migrations.ReadDir("sql")
+	if err != nil {
+		return nil, err
+	}
+
+	complete := map[string]bool{}
+	for _, mig := range allMigrations {
+		complete[mig.Name()] = false
+	}
+
+	rows, err := db.Query("select name from migrations")
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var name string
+		err = rows.Scan(&name)
+		if err != nil {
+			return nil, err
+		}
+		complete[name] = true
+	}
+
+	return complete, nil
+}
+
+// Apply a migration by name.
+func ApplyMigration(db *DB, name string) error {
+	data, err := migrations.ReadFile("sql/" + name)
+	if err != nil {
+		log.Fatalf("Missing migration %s", name)
+	}
+	log.Printf("Applying migration %s", name)
+	_, err = db.Exec(string(data))
+	if err != nil {
+		return err
+	}
+	_, err = db.Exec("insert into migrations (name) values (?)", name)
+	return err
+}
+
+// Apply all pending migrations.
+func MigrateDatabase(db *DB) error {
+	pending, err := GetPendingMigrations(db)
+	if err != nil {
+		return err
+	}
+	for name, complete := range pending {
+		if !complete {
+			err = ApplyMigration(db, name)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

core/migrations_test.go

@@ -0,0 +1,67 @@
+package core
+
+import (
+	"database/sql"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func EphemeralDb(t *testing.T) *DB {
+	// We don't use OpenDb here because you can't open two connections to the same memory mem
+	mem, err := sql.Open("sqlite3", ":memory:")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err = defaultPragma(mem); err != nil {
+		t.Fatal(err)
+	}
+	db := new(DB)
+	db.ro = mem
+	db.rw = mem
+	if err = InitDatabase(db); err != nil {
+		t.Fatal(err)
+	}
+	if err = MigrateDatabase(db); err != nil {
+		t.Fatal(err)
+	}
+	return db
+}
+
+func TestInitIdempotency(t *testing.T) {
+	mem, err := sql.Open("sqlite3", ":memory:")
+	if err != nil {
+		t.Fatal(err)
+	}
+	db := new(DB)
+	db.ro = mem
+	db.rw = mem
+	if err = InitDatabase(db); err != nil {
+		t.Fatal(err)
+	}
+	if err = InitDatabase(db); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestMigrations(t *testing.T) {
+	db := EphemeralDb(t)
+
+	allMigrations, err := migrations.ReadDir("sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rows, err := db.Query("select name from migrations")
+	if err != nil {
+		t.Fatal(err)
+	}
+	count := 0
+	for rows.Next() {
+		count += 1
+	}
+
+	if count != len(allMigrations) {
+		t.Fatalf("Expected %d migrations, got %d", len(allMigrations), count)
+	}
+}

core/source.go

@@ -0,0 +1,332 @@
+package core
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	_ "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 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, 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 *sql.Tx) 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)
+}
+
+// Given the results of a fetch, add new items, update existing items, and delete expired items.
+//
+// Returns the number of new and deleted items on success.
+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{}
+	existingItemsById := map[string]*Item{}
+	for _, item := range existingItems {
+		existingIds[item.Id] = true
+		existingItemsById[item.Id] = &item
+	}
+
+	// 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
+	for _, item := range updatedItems {
+		BackfillItem(&item, existingItemsById[item.Id])
+	}
+	if err = UpdateItems(db, updatedItems); err != nil {
+		return 0, 0, err
+	}
+
+	// If the source has an on-create trigger, run it for each new item
+	// On-create errors are ignored to avoid failing the fetch
+	onCreateArgv, err := GetArgvForAction(db, source, "on_create")
+	if err == nil {
+		var updatedNewItems []Item
+		for _, item := range newItems {
+			itemJson, err := json.Marshal(item)
+			if err != nil {
+				log.Fatalf("error: failed to serialize item: %v", err)
+			}
+			res, err := Execute(source, onCreateArgv, nil, string(itemJson), time.Minute)
+			if err != nil {
+				log.Printf("error: failed to execute on_create for %s/%s: %v", item.Source, item.Id, err)
+				continue
+			}
+			if len(res) != 1 {
+				log.Printf("error: expected on_create for %s/%s to produce exactly one item, got %d", item.Source, item.Id, len(res))
+			}
+			updatedItem := res[0]
+			BackfillItem(&updatedItem, &item)
+			updatedNewItems = append(updatedNewItems, updatedItem)
+		}
+		UpdateItems(db, updatedNewItems)
+	}
+
+	// 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
+}

core/source_test.go

@@ -0,0 +1,293 @@
+package core
+
+import (
+	"fmt"
+	"slices"
+	"testing"
+	"time"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func TestCreateSource(t *testing.T) {
+	db := EphemeralDb(t)
+
+	if err := AddSource(db, "one"); err != nil {
+		t.Fatal(err)
+	}
+	if err := AddSource(db, "two"); err != nil {
+		t.Fatal(err)
+	}
+	if err := AddSource(db, "three"); err != nil {
+		t.Fatal(err)
+	}
+	if err := DeleteSource(db, "two"); err != nil {
+		t.Fatal(err)
+	}
+
+	names, err := GetSources(db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	expected := []string{"one", "three"}
+	for i := 0; i < len(expected); i += 1 {
+		if !slices.Contains(names, expected[i]) {
+			t.Fatalf("missing %s, have: %v", expected[i], names)
+		}
+	}
+}
+
+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 {
+		t.Fatal(err)
+	}
+
+	a := Item{Source: "test", Id: "a"}
+	add, del, err := UpdateWithFetchedItems(db, "test", []Item{a})
+	if add != 1 || del != 0 || err != nil {
+		t.Fatalf("update failed: add %d, del %d, err %v", add, del, err)
+	}
+
+	add, del, err = UpdateWithFetchedItems(db, "test", []Item{a})
+	if add != 0 || del != 0 || err != nil {
+		t.Fatalf("update failed: add %d, del %d, err %v", add, del, err)
+	}
+
+	b := Item{Source: "test", Id: "b"}
+	add, del, err = UpdateWithFetchedItems(db, "test", []Item{a, b})
+	if add != 1 || del != 0 || err != nil {
+		t.Fatalf("update failed: add %d, del %d, err %v", add, del, err)
+	}
+
+	if _, err = DeactivateItem(db, "test", "a"); err != nil {
+		t.Fatal(err)
+	}
+	add, del, err = UpdateWithFetchedItems(db, "test", []Item{a, b})
+	if add != 0 || del != 0 || err != nil {
+		t.Fatalf("update failed: add %d, del %d, err %v", add, del, err)
+	}
+
+	add, del, err = UpdateWithFetchedItems(db, "test", []Item{b})
+	if add != 0 || del != 1 || err != nil {
+		t.Fatalf("update failed: add %d, del %d, err %v", add, del, err)
+	}
+
+	add, del, err = UpdateWithFetchedItems(db, "test", []Item{b})
+	if add != 0 || del != 0 || err != nil {
+		t.Fatalf("update failed: add %d, del %d, err %v", add, del, err)
+	}
+}
+
+func TestOnCreateAction(t *testing.T) {
+	db := EphemeralDb(t)
+	if err := AddSource(db, "test"); err != nil {
+		t.Fatal(err)
+	}
+	if err := AddAction(db, "test", "on_create", []string{"true"}); err != nil {
+		t.Fatal(err)
+	}
+
+	execute := func(argv []string) []Item {
+		items, err := Execute("test", argv, nil, "", time.Minute)
+		if err != nil {
+			t.Fatal("unexpected error executing test fetch")
+		}
+		if len(items) != 1 {
+			t.Fatalf("expected only one item, got %d", len(items))
+		}
+		return items
+	}
+
+	onCreate := func(argv []string) {
+		if err := UpdateAction(db, "test", "on_create", argv); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	getItem := func(id string) Item {
+		item, err := GetItem(db, "test", id)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return item
+	}
+
+	// Noop on_create works
+	onCreate([]string{"tee"})
+	items := execute([]string{"jq", "-cn", `{id: "one"}`})
+	add, _, err := UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with noop oncreate")
+	}
+	updated := getItem("one")
+	updated.Created = 0 // zero out for comparison with pre-insert item
+	if !ItemsAreEqual(updated, items[0]) {
+		t.Fatalf("expected no change: %#v != %#v", updated, items[0])
+	}
+
+	// on_create can change a field
+	onCreate([]string{"jq", "-c", `.title = "Goodbye, World"`})
+	items = execute([]string{"jq", "-cn", `{id: "two", title: "Hello, World"}`})
+	if items[0].Title != "Hello, World" {
+		t.Fatal("unexpected title")
+	}
+	add, _, err = UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with alter oncreate")
+	}
+	two := getItem("two")
+	if two.Title != "Goodbye, World" {
+		t.Fatalf("title not updated, is: %s", two.Title)
+	}
+
+	// on_create can add a field
+	onCreate([]string{"jq", "-c", `.link = "gopher://go.dev"`})
+	items = execute([]string{"jq", "-cn", `{id: "three"}`})
+	if items[0].Link != "" {
+		t.Fatal("unexpected link")
+	}
+	add, _, err = UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with augment oncreate")
+	}
+	if getItem("three").Link != "gopher://go.dev" {
+		t.Fatal("link not added")
+	}
+
+	// on_create can't delete a field using a zero value
+	// due to zero values preserving prior field values
+	onCreate([]string{"jq", "-c", `del(.link)`})
+	items = execute([]string{"jq", "-cn", `{id: "four", link: "gopher://go.dev"}`})
+	if items[0].Link != "gopher://go.dev" {
+		t.Fatal("missing link")
+	}
+	add, _, err = UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with attempted deletion oncreate")
+	}
+	if getItem("four").Link != "gopher://go.dev" {
+		t.Fatal("link unexpectedly removed")
+	}
+
+	// item is created if on_create fails
+	onCreate([]string{"false"})
+	items = execute([]string{"jq", "-cn", `{id: "five"}`})
+	add, _, err = UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with failing oncreate")
+	}
+	if getItem("five").Id != "five" {
+		t.Fatal("item not created")
+	}
+
+	// item isn't updated if on_create has valid output but a bad exit code
+	onCreate([]string{"sh", "-c", `jq -cn '{id: "six", title: "after"}'; exit 1`})
+	items = execute([]string{"jq", "-cn", `{id: "six", title: "before"}`})
+	if items[0].Title != "before" {
+		t.Fatal("unexpected title")
+	}
+	add, _, err = UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with bad exit code oncreate")
+	}
+	if getItem("six").Title != "before" {
+		t.Fatal("update applied after oncreate failed")
+	}
+
+	// on_create can't change id, active, or created
+	onCreate([]string{"jq", "-c", `.id = "seven"; .active = false; .created = 123456`})
+	items = execute([]string{"jq", "-cn", `{id: "seven"}`})
+	add, _, err = UpdateWithFetchedItems(db, "test", items)
+	if add != 1 || err != nil {
+		t.Fatal("failed update with invalid field changes oncreate")
+	}
+	updated = getItem("seven")
+	if updated.Id != "seven" || !updated.Active || updated.Created == 123456 {
+		t.Fatal("unexpected changes to id, active, or created fields")
+	}
+}

core/sql/0000_baseline.sql

@@ -0,0 +1 @@
+create table migrations (name text) strict;

core/sql/0001_initial_schema.sql

@@ -0,0 +1,25 @@
+create table sources(
+    name text not null,
+    primary key (name)
+) strict;
+create table actions(
+    source text not null,
+    name text not null,
+    argv blob not null,
+    primary key (source, name),
+    foreign key (source) references sources (name) on delete cascade
+) strict;
+create table items(
+    source text not null,
+    id text not null,
+    created int not null default (unixepoch()),
+    active int not null,
+    title text,
+    author text,
+    body text,
+    link text,
+    time int,
+    action blob,
+    primary key (source, id),
+    foreign key (source) references sources (name) on delete cascade
+) strict;

flake.lock

@@ -0,0 +1,58 @@
+{
+  "nodes": {
+    "flake-parts": {
+      "inputs": {
+        "nixpkgs-lib": "nixpkgs-lib"
+      },
+      "locked": {
+        "lastModified": 1736143030,
+        "narHash": "sha256-+hu54pAoLDEZT9pjHlqL9DNzWz0NbUn8NEAHP7PQPzU=",
+        "owner": "hercules-ci",
+        "repo": "flake-parts",
+        "rev": "b905f6fc23a9051a6e1b741e1438dbfc0634c6de",
+        "type": "github"
+      },
+      "original": {
+        "owner": "hercules-ci",
+        "repo": "flake-parts",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1736798957,
+        "narHash": "sha256-qwpCtZhSsSNQtK4xYGzMiyEDhkNzOCz/Vfu4oL2ETsQ=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "9abb87b552b7f55ac8916b6fc9e5cb486656a2f3",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixos-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "nixpkgs-lib": {
+      "locked": {
+        "lastModified": 1735774519,
+        "narHash": "sha256-CewEm1o2eVAnoqb6Ml+Qi9Gg/EfNAxbRx1lANGVyoLI=",
+        "type": "tarball",
+        "url": "https://github.com/NixOS/nixpkgs/archive/e9b51731911566bbf7e4895475a87fe06961de0b.tar.gz"
+      },
+      "original": {
+        "type": "tarball",
+        "url": "https://github.com/NixOS/nixpkgs/archive/e9b51731911566bbf7e4895475a87fe06961de0b.tar.gz"
+      }
+    },
+    "root": {
+      "inputs": {
+        "flake-parts": "flake-parts",
+        "nixpkgs": "nixpkgs"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,40 @@
+{
+  description = "Universal and extensible feed aggregator";
+
+  inputs = {
+    flake-parts.url = "github:hercules-ci/flake-parts";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+  };
+
+  outputs =
+    inputs@{ flake-parts, ... }:
+    flake-parts.lib.mkFlake { inherit inputs; } {
+      systems = [
+        "x86_64-linux"
+        "aarch64-linux"
+        "aarch64-darwin"
+        "x86_64-darwin"
+      ];
+      perSystem =
+        {
+          pkgs,
+          ...
+        }:
+        {
+          formatter = pkgs.nixfmt-rfc-style;
+
+          devShells.default = pkgs.mkShell {
+            packages = [
+              pkgs.go
+              pkgs.gopls
+              pkgs.go-tools
+              pkgs.gotools
+              pkgs.cobra-cli
+              pkgs.air
+            ];
+          };
+        };
+      flake = {
+      };
+    };
+}

go.mod

@@ -0,0 +1,12 @@
+module github.com/Jaculabilis/intake
+
+go 1.23.4
+
+require github.com/spf13/cobra v1.8.1
+
+require github.com/mattn/go-sqlite3 v1.14.24
+
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+)

go.sum

@@ -0,0 +1,12 @@
+github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/mattn/go-sqlite3 v1.14.24 h1:tpSp2G2KyMnnQu99ngJ47EIkWVmliIizyZBfPrBWDRM=
+github.com/mattn/go-sqlite3 v1.14.24/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
+github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y=
+github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

main.go

@@ -0,0 +1,7 @@
+package main
+
+import "github.com/Jaculabilis/intake/cmd"
+
+func main() {
+	cmd.Execute()
+}

test/test_items.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+set -eu
+
+go build -o tmp/intake
+rm tmp/intake.db* || true
+export INTAKE_DATA_DIR="tmp"
+tmp/intake migrate
+
+tmp/intake source add -s feedtest
+tmp/intake item add -s feedtest --id "this-item-has-no-title"
+tmp/intake item add -s feedtest --title "This item has only a title"
+tmp/intake item add -s feedtest --title "Title and body" --body "This is the item body"
+tmp/intake item add -s feedtest --title "Title and link" --link "#"
+tmp/intake item add -s feedtest --title "Title, link, body" --link "#" --body "This is the body"
+tmp/intake item add -s feedtest --title "<b>HTML title</b>" --link "#" --body "<i>HTML body</i>"
+tmp/intake item add -s feedtest --title "Title and author" --author "Authorname"
+tmp/intake item add -s feedtest --title "Title, author, time" --author "Authorname" --time 1700000000
+tmp/intake item add -s feedtest --title "Title, time" --time 1737780324
+tmp/intake item add -s feedtest --title "Title, author, body" --author "Authorname" --body "Hello body!"
+tmp/intake item add -s feedtest --title "Title, author, time, body" --author "Authorname" --time 1700000000 --body "Hello body!"
+tmp/intake item add -s feedtest --title "Title, time, body" --time 1737780324 --body "Hello, body!"
+
+tmp/intake source add -s spook
+tmp/intake action add -s spook -a spookier -- jq -c '.title = .title + "o"'
+tmp/intake item add -s spook --id boo --title "Boo" --action '{"spookier": true}'

web/html/feed.html

@@ -0,0 +1,32 @@
+{{ define "title" }}{{ if .Items }}({{ len .Items }}) {{ end }}Intake{{ end }}
+
+{{ define "content" -}}
+<article class="center">
+<span class="item-title">
+<a href="/">Home</a>
+[<a href="#">Active</a> | <a href="#">All</a>]
+</span>
+</article>
+
+{{ if .Items }}
+{{ range .Items }}
+	{{ template "item" . }}
+{{ end }}
+
+<article class="center">
+<button
+    hx-post="/mass-deactivate"
+    hx-vals='{{ massDeacVars .Items }}'
+    hx-confirm="Deactivate {{ len .Items }} items?"
+>Deactivate All</button>
+</article>
+
+{{ else }}
+<article class="center">
+    <span class="item-title">Feed is empty</span>
+</article>
+{{ end }}
+{{/* end if .Items */}}
+
+{{ end }}
+{{/* end define "content" */}}

web/html/home.html

@@ -0,0 +1,20 @@
+{{ define "title" }}Intake{{ end }}
+
+{{ define "content" -}}
+<article>
+<details>
+<summary><span class="item-title">Sources</span></summary>
+{{ if .Sources }}
+<table class="intake-sources">
+{{ range .Sources }}
+<tr>
+<td><a href="/source/{{ .Name }}">{{ .Name }}</a></td>
+</tr>
+{{ end }}
+</table>
+{{ else }}
+<p>No sources found.</p>
+{{ end }}
+</details>
+</article>
+{{- end }}

web/html/html.go

@@ -0,0 +1,91 @@
+package html
+
+import (
+	"embed"
+	"encoding/json"
+	"html/template"
+	"io"
+	"log"
+	"time"
+
+	"github.com/Jaculabilis/intake/core"
+)
+
+func rawHtml(str string) template.HTML {
+	return template.HTML(str)
+}
+
+func tsToDate(t int) string {
+	tm := time.Unix(int64(t), 0).UTC()
+	return tm.Format(time.DateTime)
+}
+
+func massDeactivateVals(items []core.Item) string {
+	var shorts []string
+	for _, item := range items {
+		shorts = append(shorts, core.FormatAsShort(item))
+	}
+	massDeac := struct {
+		Items []string `json:"items"`
+	}{shorts}
+	vals, err := json.Marshal(massDeac)
+	if err != nil {
+		log.Printf("error serializing mass deactivate list: %v", err)
+	}
+	return string(vals)
+}
+
+var funcs = template.FuncMap{
+	"raw":          rawHtml,
+	"tsToDate":     tsToDate,
+	"massDeacVars": massDeactivateVals,
+}
+
+//go:embed intake.css
+var Stylesheet []byte
+
+//go:embed htmx.org@2.0.4.js
+var Htmx []byte
+
+//go:embed *.html
+var templates embed.FS
+
+func load(files ...string) *template.Template {
+	files = append([]string{"layout.html"}, files...)
+	return template.Must(template.New("layout.html").Funcs(funcs).ParseFS(templates, files...))
+}
+
+var home = load("home.html")
+
+type SourceData struct {
+	Name string
+}
+
+type HomeData struct {
+	Sources []SourceData
+}
+
+func Home(writer io.Writer, data HomeData) error {
+	return home.Execute(writer, data)
+}
+
+var feed = load("feed.html", "item.html")
+
+type FeedData struct {
+	Items []core.Item
+}
+
+func Feed(writer io.Writer, data FeedData) error {
+	return feed.Execute(writer, data)
+}
+
+var item = load("itemPage.html", "item.html")
+
+type ItemData struct {
+	Item core.Item
+	Open bool
+}
+
+func Item(writer io.Writer, data ItemData) error {
+	return item.Execute(writer, data)
+}

web/html/intake.css

@@ -0,0 +1,85 @@
+
+main {
+	max-width: 700px;
+	margin: 0 auto;
+}
+article {
+	border: 1px solid black; border-radius: 6px;
+	padding: 5px;
+	margin-bottom: 20px;
+	word-break: break-word;
+	display: flow-root;
+}
+.item-title {
+	font-size: 1.4em;
+}
+.item-button {
+	font-size: 1em;
+	float:right;
+	margin-left: 2px;
+}
+.item-link {
+	text-decoration: none;
+	float:right;
+	font-size: 1em;
+	padding: 2px 7px;
+	border: 1px solid;
+	border-radius: 2px;
+}
+.item-info {
+	opacity: 0.7;
+}
+details[open] > summary > .item-button, details[open] > summary > .item-link {
+	display: none;
+}
+details ~ .item-button, details ~ .item-link {
+	display: none;
+}
+details[open] ~ .item-button, details[open] ~ .item-link {
+	display: inline;
+}
+article img {
+	max-width: 100%;
+	height: auto;
+}
+button, summary {
+	cursor: pointer;
+}
+summary {
+	display: block;
+}
+summary:focus {
+	outline: 1px dotted gray;
+}
+.strikethru span, .strikethru p {
+	text-decoration: line-through;
+}
+.wide {
+	width: 100%;
+	resize: vertical;
+}
+.fade > * {
+	opacity: 0.2;
+}
+pre {
+	white-space: pre-wrap;
+}
+table.feed-control td {
+	font-family: monospace; padding: 5px 10px;
+}
+.intake-sources td {
+	padding-block: 0.4em;
+}
+.intake-sources form {
+	margin: 0
+}
+article.center {
+	text-align: center;
+}
+article textarea {
+	width: 100%;
+	resize: vertical;
+}
+span.error-message {
+	color: red;
+}

web/html/item.html

@@ -0,0 +1,70 @@
+{{ define "item-buttons" -}}
+<button
+	class="item-button"
+	title="Deactivate {{ .Source }}/{{ .Id }}"
+	hx-target="closest article"
+	hx-select="article"
+	hx-delete="/item/{{ .Source }}/{{ .Id }}"
+>&#10005;</button>
+<button
+	class="item-button"
+	title="Punt {{ .Source }}/{{ .Id }}"
+>&#8631;</button>
+{{- if .Link }}<a class="item-link" href="{{ .Link }}" target="_blank">&#8663;</a>
+{{ end -}}
+{{ range $key, $_ := .Action }}
+<button
+    class="item-button"
+    title="{{ $key }}"
+    hx-target="closest article"
+    hx-select="article"
+    hx-disabled-elt="this"
+    hx-post="/item/{{ $.Source }}/{{ $.Id }}/action/{{ $key }}"
+>{{ $key }}</button>
+{{ end -}}
+{{ end }}
+
+{{ define "item-title" -}}
+<span class="item-title">{{ or .Title .Id | raw }}</span>
+{{- end }}
+
+{{ define "item-class" -}}{{ if not .Active }}strikethru {{ end }}{{ if not .Active }}fade{{ end }}{{- end}}
+
+{{ define "item" -}}
+<article
+	id="{{ .Source }}-{{ .Id }}"
+	class="{{ template "item-class" . }}"
+>
+
+{{- /* The item title is a clickable <summary> if there is body content */ -}}
+{{ if .Body }}
+<details>
+<summary>
+    {{ template "item-buttons" . }}
+    {{ template "item-title" . }}
+</summary>
+<p>{{ raw .Body }}</p>
+</details>
+{{ template "item-buttons" . }}
+{{- else -}}
+{{ template "item-buttons" . }}
+{{ template "item-title" . }}<br>
+{{ end }}
+{{- /* end if .Body */ -}}
+
+{{- /* author/time footer line */ -}}
+{{ if or .Author .Time }}
+<span class="item-info">
+{{ .Author }}
+{{ .Time | tsToDate }}
+</span><br>
+{{ end -}}
+
+{{- /* source/id/created footer line */ -}}
+<span class="item-info">
+<a href="/item/{{ .Source }}/{{ .Id }}">{{ .Source }}/{{ .Id }}</a>
+{{ .Created | tsToDate }}
+</span>
+</article>
+{{ end -}}
+{{- /* end define "item" */ -}}

web/html/itemPage.html

@@ -0,0 +1,5 @@
+{{ define "title" }}{{ if .Item.Title }}{{ .Item.Title }}{{ else }}{{ .Item.Source }}/{{ .Item.Id }}{{ end }} - Intake [{{ .Item.Source }}]{{ end }}
+
+{{ define "content" -}}
+{{ template "item" .Item }}
+{{- end }}

web/html/layout.html

@@ -0,0 +1,16 @@
+<!DOCTYPE html>
+<html>
+<head>
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>{{ block "title" . }}Intake{{ end }}</title>
+<link rel="icon" type="image/png" href="data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAABAAAAAQCAYAAAAf8/9hAAAABGdBTUEAALGPC/xhBQAAAAlwSFlzAAAOwgAADsIBFShKgAAAABh0RVh0U29mdHdhcmUAcGFpbnQubmV0IDQuMS41ZEdYUgAAAGFJREFUOE+lkFEKwDAIxXrzXXB3ckMm9EnAV/YRCxFCcUXEL3Jc77NDjpDA/VGL3RFWYEICfeGC8oQc9IPuCAnQDcoRVmBCAn3hgvKEHPSD7ggJ0A3KEVZgQgJ94YLSJ9YDUzNGDXGZ/JEAAAAASUVORK5CYII=">
+<link rel="stylesheet" href="/style.css">
+<script src="/htmx.org@2.0.4.js"></script>
+<meta name="htmx-config" content='{"ignoreTitle":true,"defaultSwapStyle":"outerHTML"}'>
+</head>
+<body>
+<main>
+{{ template "content" . }}
+</main>
+</body>
+</html>

web/item.go

@@ -0,0 +1,119 @@
+package web
+
+import (
+	"encoding/json"
+	"log"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/Jaculabilis/intake/web/html"
+)
+
+func (env *Env) getItem(writer http.ResponseWriter, req *http.Request) {
+	source := req.PathValue("source")
+	id := req.PathValue("id")
+
+	item, err := core.GetItem(env.db, source, id)
+	if err != nil {
+		writer.Write([]byte(err.Error()))
+		return
+	}
+	html.Item(writer, html.ItemData{Item: item})
+}
+
+func (env *Env) deleteItem(writer http.ResponseWriter, req *http.Request) {
+	source := req.PathValue("source")
+	id := req.PathValue("id")
+
+	_, err := core.DeactivateItem(env.db, source, id)
+	if err != nil {
+		writer.Write([]byte(err.Error()))
+		return
+	}
+	item, err := core.GetItem(env.db, source, id)
+	if err != nil {
+		writer.Write([]byte(err.Error()))
+		return
+	}
+	html.Item(writer, html.ItemData{Item: item})
+}
+
+func (env *Env) doAction(writer http.ResponseWriter, req *http.Request) {
+	source := req.PathValue("source")
+	id := req.PathValue("id")
+	action := req.PathValue("action")
+
+	item, err := core.GetItem(env.db, source, id)
+	if err != nil {
+		http.Error(writer, err.Error(), 500)
+		return
+	}
+
+	if item.Action[action] == nil {
+		http.Error(writer, "no such action", 500)
+		return
+	}
+
+	argv, err := core.GetArgvForAction(env.db, source, action)
+	if err != nil {
+		http.Error(writer, err.Error(), 500)
+		return
+	}
+
+	itemJson, err := json.Marshal(item)
+	if err != nil {
+		http.Error(writer, err.Error(), 500)
+		return
+	}
+
+	res, err := core.Execute(source, argv, nil, string(itemJson), time.Minute)
+	if err != nil {
+		http.Error(writer, err.Error(), 500)
+		return
+	}
+	if len(res) != 1 {
+		http.Error(writer, "not exactly one item", 500)
+		return
+	}
+	newItem := res[0]
+	core.BackfillItem(&newItem, &item)
+
+	if err = core.UpdateItems(env.db, []core.Item{newItem}); err != nil {
+		http.Error(writer, err.Error(), 500)
+		return
+	}
+
+	html.Item(writer, html.ItemData{Item: newItem})
+}
+
+func (env *Env) massDeactivate(writer http.ResponseWriter, req *http.Request) {
+	if err := req.ParseForm(); err != nil {
+		log.Printf("error parsing form data: %v", err)
+		http.Error(writer, "", http.StatusBadRequest)
+		return
+	}
+	for _, item := range req.PostForm["items"] {
+		i := strings.Index(item, "/")
+		if i == -1 {
+			log.Printf("error: invalid source/item: %s", item)
+			http.Error(writer, "", http.StatusBadRequest)
+			return
+		}
+	}
+	for _, item := range req.PostForm["items"] {
+		i := strings.Index(item, "/")
+		source := item[:i]
+		id := item[i+1:]
+		active, err := core.DeactivateItem(env.db, source, id)
+		if err != nil {
+			log.Printf("error: failed to deactivate %s/%s: %v", source, id, err)
+		}
+		if active {
+			log.Printf("deactivated %s/%s", source, id)
+		}
+	}
+	writer.Header()["HX-Refresh"] = []string{"true"}
+	http.Error(writer, "ok", http.StatusNoContent)
+}

web/main.go

@@ -0,0 +1,40 @@
+package web
+
+import (
+	"log"
+	"net"
+	"net/http"
+
+	"github.com/Jaculabilis/intake/core"
+)
+
+type Env struct {
+	db *core.DB
+}
+
+func logged(handler http.HandlerFunc) http.HandlerFunc {
+	return func(writer http.ResponseWriter, req *http.Request) {
+		log.Printf("%s %s", req.Method, req.URL.Path)
+		handler(writer, req)
+	}
+}
+
+func handleFunc(pattern string, handler http.HandlerFunc) {
+	http.HandleFunc(pattern, logged(handler))
+}
+
+func RunServer(db *core.DB, addr string, port string) {
+	env := &Env{db}
+	bind := net.JoinHostPort(addr, port)
+
+	handleFunc("GET /", env.getRoot)
+	handleFunc("GET /style.css", env.getStyle)
+	handleFunc("GET /htmx.org@2.0.4.js", env.getScript)
+	handleFunc("GET /source/{source}", env.getSource)
+	handleFunc("GET /item/{source}/{id}", env.getItem)
+	handleFunc("DELETE /item/{source}/{id}", env.deleteItem)
+	handleFunc("POST   /item/{source}/{id}/action/{action}", env.doAction)
+	handleFunc("POST   /mass-deactivate", env.massDeactivate)
+
+	log.Fatal(http.ListenAndServe(bind, nil))
+}

web/root.go

@@ -0,0 +1,41 @@
+package web
+
+import (
+	"net/http"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/Jaculabilis/intake/web/html"
+)
+
+func (env *Env) getRoot(writer http.ResponseWriter, req *http.Request) {
+	if req.URL.Path != "/" {
+		http.NotFound(writer, req)
+		return
+	}
+
+	names, err := core.GetSources(env.db)
+	if err != nil {
+		writer.Write([]byte(err.Error()))
+	}
+
+	var sources []html.SourceData
+	for _, name := range names {
+		sources = append(sources, html.SourceData{Name: name})
+	}
+	data := html.HomeData{
+		Sources: sources,
+	}
+	html.Home(writer, data)
+}
+
+func (env *Env) getStyle(writer http.ResponseWriter, req *http.Request) {
+	writer.Header()["Cache-Control"] = []string{"public, max-age=86400"}
+	writer.Header()["Content-Type"] = []string{"text/css; charset=utf-8"}
+	writer.Write(html.Stylesheet)
+}
+
+func (env *Env) getScript(writer http.ResponseWriter, req *http.Request) {
+	writer.Header()["Cache-Control"] = []string{"public, max-age=86400"}
+	writer.Header()["Content-Type"] = []string{"application/javascript; charset=utf-8"}
+	writer.Write(html.Htmx)
+}

web/source.go

@@ -0,0 +1,27 @@
+package web
+
+import (
+	"net/http"
+
+	"github.com/Jaculabilis/intake/core"
+	"github.com/Jaculabilis/intake/web/html"
+)
+
+func (env *Env) getSource(writer http.ResponseWriter, req *http.Request) {
+	source := req.PathValue("source")
+	if source == "" {
+		http.NotFound(writer, req)
+		return
+	}
+
+	// TODO this needs to properly error if the source doesn't exist instead of just returning []
+	items, err := core.GetAllItemsForSource(env.db, source)
+	if err != nil {
+		http.NotFound(writer, req)
+		return
+	}
+	data := html.FeedData{
+		Items: items,
+	}
+	html.Feed(writer, data)
+}