Compare commits
8 Commits
a0e4f1f9c4
...
8940fdf697
| Author | SHA1 | Date | |
|---|---|---|---|
| 8940fdf697 | |||
| f231c81f2d | |||
| 0337cc8ee3 | |||
| c9949b7b25 | |||
| 79327bb34a | |||
| 9132bb46d7 | |||
| e695094a0a | |||
| 860c8008e0 |
180
core/action.go
180
core/action.go
@ -1,18 +1,8 @@
|
|||||||
package core
|
package core
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bufio"
|
|
||||||
"context"
|
|
||||||
"database/sql/driver"
|
"database/sql/driver"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"log"
|
|
||||||
"os"
|
|
||||||
"os/exec"
|
|
||||||
"strings"
|
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// Type alias for storing string array as jsonb
|
// Type alias for storing string array as jsonb
|
||||||
@ -85,173 +75,3 @@ func DeleteAction(db DB, source string, name string) error {
|
|||||||
`, source, name)
|
`, source, name)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func readStdout(stdout io.ReadCloser, source string, items chan Item, cparse chan bool) {
|
|
||||||
var item Item
|
|
||||||
parseError := false
|
|
||||||
scanout := bufio.NewScanner(stdout)
|
|
||||||
for scanout.Scan() {
|
|
||||||
data := scanout.Bytes()
|
|
||||||
err := json.Unmarshal(data, &item)
|
|
||||||
if err != nil || item.Id == "" {
|
|
||||||
log.Printf("[%s: stdout] %s\n", source, strings.TrimSpace(string(data)))
|
|
||||||
parseError = true
|
|
||||||
} else {
|
|
||||||
item.Active = true // These fields aren't up to
|
|
||||||
item.Created = 0 // the action to set and
|
|
||||||
item.Source = source // shouldn't be overrideable
|
|
||||||
log.Printf("[%s: item] %s\n", source, item.Id)
|
|
||||||
items <- item
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Only send the parsing result at the end, to block main until stdout is drained
|
|
||||||
cparse <- parseError
|
|
||||||
close(items)
|
|
||||||
}
|
|
||||||
|
|
||||||
func readStderr(stderr io.ReadCloser, source string, done chan bool) {
|
|
||||||
scanerr := bufio.NewScanner(stderr)
|
|
||||||
for scanerr.Scan() {
|
|
||||||
text := strings.TrimSpace(scanerr.Text())
|
|
||||||
log.Printf("[%s: stderr] %s\n", source, text)
|
|
||||||
}
|
|
||||||
done <- true
|
|
||||||
}
|
|
||||||
|
|
||||||
func writeStdin(stdin io.WriteCloser, text string) {
|
|
||||||
defer stdin.Close()
|
|
||||||
io.WriteString(stdin, text)
|
|
||||||
}
|
|
||||||
|
|
||||||
func Execute(
|
|
||||||
source string,
|
|
||||||
argv []string,
|
|
||||||
env []string,
|
|
||||||
state []byte,
|
|
||||||
input string,
|
|
||||||
timeout time.Duration,
|
|
||||||
) ([]Item, []byte, error) {
|
|
||||||
log.Printf("executing %v", argv)
|
|
||||||
|
|
||||||
if len(argv) == 0 {
|
|
||||||
return nil, nil, errors.New("empty argv")
|
|
||||||
}
|
|
||||||
if source == "" {
|
|
||||||
return nil, nil, errors.New("empty source")
|
|
||||||
}
|
|
||||||
|
|
||||||
stateFile, err := os.CreateTemp("", "intake_state_*")
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, fmt.Errorf("error: failed to create temp state file: %v", err)
|
|
||||||
}
|
|
||||||
defer func() {
|
|
||||||
if err := os.Remove(stateFile.Name()); err != nil {
|
|
||||||
log.Printf("error: failed to delete %s", stateFile.Name())
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
_, err = stateFile.Write(state)
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, fmt.Errorf("error: failed to write state file: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
env = append(env, "STATE_PATH="+stateFile.Name())
|
|
||||||
|
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
||||||
defer cancel()
|
|
||||||
cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
|
|
||||||
cmd.Env = append(os.Environ(), env...)
|
|
||||||
cmd.WaitDelay = time.Second * 5
|
|
||||||
|
|
||||||
// Open pipes to the command
|
|
||||||
stdin, err := cmd.StdinPipe()
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
stdout, err := cmd.StdoutPipe()
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
stderr, err := cmd.StderrPipe()
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
cout := make(chan Item)
|
|
||||||
cparse := make(chan bool)
|
|
||||||
cerr := make(chan bool)
|
|
||||||
|
|
||||||
// Sink routine for items produced
|
|
||||||
var items []Item
|
|
||||||
go func() {
|
|
||||||
for item := range cout {
|
|
||||||
items = append(items, item)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Routines handling the process i/o
|
|
||||||
go writeStdin(stdin, input)
|
|
||||||
go readStdout(stdout, source, cout, cparse)
|
|
||||||
go readStderr(stderr, source, cerr)
|
|
||||||
|
|
||||||
// Kick off the command
|
|
||||||
err = cmd.Start()
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Block until std{out,err} close
|
|
||||||
<-cerr
|
|
||||||
parseError := <-cparse
|
|
||||||
|
|
||||||
err = cmd.Wait()
|
|
||||||
if ctx.Err() == context.DeadlineExceeded {
|
|
||||||
log.Printf("Timed out after %v\n", timeout)
|
|
||||||
return nil, nil, err
|
|
||||||
} else if exiterr, ok := err.(*exec.ExitError); ok {
|
|
||||||
log.Printf("error: %s failed with exit code %d\n", argv[0], exiterr.ExitCode())
|
|
||||||
return nil, nil, err
|
|
||||||
} else if err != nil {
|
|
||||||
log.Printf("error: %s failed with error: %s\n", argv[0], err)
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
if parseError {
|
|
||||||
log.Printf("error: could not parse item\n")
|
|
||||||
return nil, nil, errors.New("invalid JSON")
|
|
||||||
}
|
|
||||||
|
|
||||||
newState, err := os.ReadFile(stateFile.Name())
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, fmt.Errorf("error: failed to read state file: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return items, newState, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Execute an action that takes an item as input and returns the item modified.
|
|
||||||
// This is basically just a wrapper over [Execute] that handles the input and backfilling.
|
|
||||||
func ExecuteItemAction(
|
|
||||||
item Item,
|
|
||||||
argv []string,
|
|
||||||
env []string,
|
|
||||||
state []byte,
|
|
||||||
timeout time.Duration,
|
|
||||||
) (Item, []byte, error) {
|
|
||||||
itemJson, err := json.Marshal(item)
|
|
||||||
if err != nil {
|
|
||||||
return Item{}, nil, fmt.Errorf("failed to serialize item: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
res, newState, err := Execute(item.Source, argv, env, state, string(itemJson), timeout)
|
|
||||||
if err != nil {
|
|
||||||
return Item{}, nil, fmt.Errorf("failed to execute action for %s/%s: %v", item.Source, item.Id, err)
|
|
||||||
}
|
|
||||||
if len(res) != 1 {
|
|
||||||
return Item{}, nil, fmt.Errorf("expected action to produce exactly one item, got %d", len(res))
|
|
||||||
}
|
|
||||||
newItem := res[0]
|
|
||||||
BackfillItem(&newItem, &item)
|
|
||||||
|
|
||||||
return newItem, newState, nil
|
|
||||||
}
|
|
||||||
|
|||||||
@ -2,7 +2,6 @@ package core
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestActionCreate(t *testing.T) {
|
func TestActionCreate(t *testing.T) {
|
||||||
@ -54,148 +53,3 @@ func TestActionCreate(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestExecute(t *testing.T) {
|
|
||||||
assertLen := func(items []Item, length int) {
|
|
||||||
if len(items) != length {
|
|
||||||
t.Fatalf("Expected %d items, got %d", length, len(items))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
assertNil := func(err error) {
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
assertNotNil := func(err error) {
|
|
||||||
if err == nil {
|
|
||||||
t.Fatal("expected err")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
execute := func(argv []string) ([]Item, error) {
|
|
||||||
item, _, err := Execute("_", argv, nil, nil, "", time.Minute)
|
|
||||||
return item, err
|
|
||||||
}
|
|
||||||
|
|
||||||
res, err := execute([]string{"true"})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
// Exit with error code
|
|
||||||
res, err = execute([]string{"false"})
|
|
||||||
assertNotNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
res, err = execute([]string{"sh", "-c", "exit 22"})
|
|
||||||
assertNotNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
// Timeout
|
|
||||||
res, _, err = Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond)
|
|
||||||
assertNotNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
// Returning items
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "foo"}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Id != "foo" {
|
|
||||||
t.Fatal("jq -cn test failed")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Read from stdin
|
|
||||||
res, _, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute)
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Id != "bar" {
|
|
||||||
t.Fatal("jq -cR test failed")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Set env
|
|
||||||
res, _, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute)
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Id != "baz" {
|
|
||||||
t.Fatal("jq -cn env test failed")
|
|
||||||
}
|
|
||||||
|
|
||||||
// With logging on stderr
|
|
||||||
res, err = execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Id != "box" {
|
|
||||||
t.Fatal("stderr test failed")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Unsupported item field is silently discarded
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
|
|
||||||
// Field with incorrect type fails
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: ["list"]}`})
|
|
||||||
assertNotNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
|
|
||||||
assertNotNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: null}`})
|
|
||||||
assertNotNil(err)
|
|
||||||
assertLen(res, 0)
|
|
||||||
|
|
||||||
// Items with duplicate ids is not a fetch error, but it will fail to update
|
|
||||||
res, err = execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 2)
|
|
||||||
|
|
||||||
// Action keys are detected even with empty values
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Action["hello"] == nil {
|
|
||||||
t.Fatal("missing hello action")
|
|
||||||
}
|
|
||||||
if res[0].Action["goodbye"] != nil {
|
|
||||||
t.Fatal("nonexistent action should key to nil in Action")
|
|
||||||
}
|
|
||||||
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Action["hello"] == nil {
|
|
||||||
t.Fatal("missing hello action")
|
|
||||||
}
|
|
||||||
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Action["hello"] == nil {
|
|
||||||
t.Fatal("missing hello action")
|
|
||||||
}
|
|
||||||
|
|
||||||
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Action["hello"] == nil {
|
|
||||||
t.Fatal("missing hello action")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Read state
|
|
||||||
argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
|
|
||||||
res, _, err = Execute("_", argv, nil, []byte("world"), "", time.Minute)
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if res[0].Title != "Hello world" {
|
|
||||||
t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Write state
|
|
||||||
argv = []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
|
|
||||||
res, newState, err := Execute("_", argv, nil, nil, "", time.Minute)
|
|
||||||
assertNil(err)
|
|
||||||
assertLen(res, 1)
|
|
||||||
if string(newState) != "Hello world" {
|
|
||||||
t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
185
core/execute.go
Normal file
185
core/execute.go
Normal file
@ -0,0 +1,185 @@
|
|||||||
|
package core
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bufio"
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"log"
|
||||||
|
"os"
|
||||||
|
"os/exec"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func readStdout(stdout io.ReadCloser, source string, items chan Item, cparse chan bool) {
|
||||||
|
var item Item
|
||||||
|
parseError := false
|
||||||
|
scanout := bufio.NewScanner(stdout)
|
||||||
|
for scanout.Scan() {
|
||||||
|
data := scanout.Bytes()
|
||||||
|
err := json.Unmarshal(data, &item)
|
||||||
|
if err != nil || item.Id == "" {
|
||||||
|
log.Printf("[%s: stdout] %s\n", source, strings.TrimSpace(string(data)))
|
||||||
|
parseError = true
|
||||||
|
} else {
|
||||||
|
item.Active = true // These fields aren't up to
|
||||||
|
item.Created = 0 // the action to set and
|
||||||
|
item.Source = source // shouldn't be overrideable
|
||||||
|
log.Printf("[%s: item] %s\n", source, item.Id)
|
||||||
|
items <- item
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Only send the parsing result at the end, to block main until stdout is drained
|
||||||
|
cparse <- parseError
|
||||||
|
close(items)
|
||||||
|
}
|
||||||
|
|
||||||
|
func readStderr(stderr io.ReadCloser, source string, done chan bool) {
|
||||||
|
scanerr := bufio.NewScanner(stderr)
|
||||||
|
for scanerr.Scan() {
|
||||||
|
text := strings.TrimSpace(scanerr.Text())
|
||||||
|
log.Printf("[%s: stderr] %s\n", source, text)
|
||||||
|
}
|
||||||
|
done <- true
|
||||||
|
}
|
||||||
|
|
||||||
|
func writeStdin(stdin io.WriteCloser, text string) {
|
||||||
|
defer stdin.Close()
|
||||||
|
io.WriteString(stdin, text)
|
||||||
|
}
|
||||||
|
|
||||||
|
func Execute(
|
||||||
|
source string,
|
||||||
|
argv []string,
|
||||||
|
env []string,
|
||||||
|
state []byte,
|
||||||
|
input string,
|
||||||
|
timeout time.Duration,
|
||||||
|
) ([]Item, []byte, error) {
|
||||||
|
log.Printf("executing %v", argv)
|
||||||
|
|
||||||
|
if len(argv) == 0 {
|
||||||
|
return nil, nil, errors.New("empty argv")
|
||||||
|
}
|
||||||
|
if source == "" {
|
||||||
|
return nil, nil, errors.New("empty source")
|
||||||
|
}
|
||||||
|
|
||||||
|
stateFile, err := os.CreateTemp("", "intake_state_*")
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, fmt.Errorf("error: failed to create temp state file: %v", err)
|
||||||
|
}
|
||||||
|
defer func() {
|
||||||
|
if err := os.Remove(stateFile.Name()); err != nil {
|
||||||
|
log.Printf("error: failed to delete %s", stateFile.Name())
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
_, err = stateFile.Write(state)
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, fmt.Errorf("error: failed to write state file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
env = append(env, "STATE_PATH="+stateFile.Name())
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||||
|
defer cancel()
|
||||||
|
cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
|
||||||
|
cmd.Env = append(os.Environ(), env...)
|
||||||
|
cmd.WaitDelay = time.Second * 5
|
||||||
|
|
||||||
|
// Open pipes to the command
|
||||||
|
stdin, err := cmd.StdinPipe()
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
stdout, err := cmd.StdoutPipe()
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
stderr, err := cmd.StderrPipe()
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
cout := make(chan Item)
|
||||||
|
cparse := make(chan bool)
|
||||||
|
cerr := make(chan bool)
|
||||||
|
|
||||||
|
// Sink routine for items produced
|
||||||
|
var items []Item
|
||||||
|
go func() {
|
||||||
|
for item := range cout {
|
||||||
|
items = append(items, item)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
// Routines handling the process i/o
|
||||||
|
go writeStdin(stdin, input)
|
||||||
|
go readStdout(stdout, source, cout, cparse)
|
||||||
|
go readStderr(stderr, source, cerr)
|
||||||
|
|
||||||
|
// Kick off the command
|
||||||
|
err = cmd.Start()
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Block until std{out,err} close
|
||||||
|
<-cerr
|
||||||
|
parseError := <-cparse
|
||||||
|
|
||||||
|
err = cmd.Wait()
|
||||||
|
if ctx.Err() == context.DeadlineExceeded {
|
||||||
|
log.Printf("Timed out after %v\n", timeout)
|
||||||
|
return nil, nil, err
|
||||||
|
} else if exiterr, ok := err.(*exec.ExitError); ok {
|
||||||
|
log.Printf("error: %s failed with exit code %d\n", argv[0], exiterr.ExitCode())
|
||||||
|
return nil, nil, err
|
||||||
|
} else if err != nil {
|
||||||
|
log.Printf("error: %s failed with error: %s\n", argv[0], err)
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
if parseError {
|
||||||
|
log.Printf("error: could not parse item\n")
|
||||||
|
return nil, nil, errors.New("invalid JSON")
|
||||||
|
}
|
||||||
|
|
||||||
|
newState, err := os.ReadFile(stateFile.Name())
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, fmt.Errorf("error: failed to read state file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return items, newState, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Execute an action that takes an item as input and returns the item modified.
|
||||||
|
// This is basically just a wrapper over [Execute] that handles the input and backfilling.
|
||||||
|
func ExecuteItemAction(
|
||||||
|
item Item,
|
||||||
|
argv []string,
|
||||||
|
env []string,
|
||||||
|
state []byte,
|
||||||
|
timeout time.Duration,
|
||||||
|
) (Item, []byte, error) {
|
||||||
|
itemJson, err := json.Marshal(item)
|
||||||
|
if err != nil {
|
||||||
|
return Item{}, nil, fmt.Errorf("failed to serialize item: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
res, newState, err := Execute(item.Source, argv, env, state, string(itemJson), timeout)
|
||||||
|
if err != nil {
|
||||||
|
return Item{}, nil, fmt.Errorf("failed to execute action for %s/%s: %v", item.Source, item.Id, err)
|
||||||
|
}
|
||||||
|
if len(res) != 1 {
|
||||||
|
return Item{}, nil, fmt.Errorf("expected action to produce exactly one item, got %d", len(res))
|
||||||
|
}
|
||||||
|
newItem := res[0]
|
||||||
|
BackfillItem(&newItem, &item)
|
||||||
|
|
||||||
|
return newItem, newState, nil
|
||||||
|
}
|
||||||
151
core/execute_test.go
Normal file
151
core/execute_test.go
Normal file
@ -0,0 +1,151 @@
|
|||||||
|
package core
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestExecute(t *testing.T) {
|
||||||
|
assertLen := func(items []Item, length int) {
|
||||||
|
if len(items) != length {
|
||||||
|
t.Fatalf("Expected %d items, got %d", length, len(items))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assertNil := func(err error) {
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assertNotNil := func(err error) {
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected err")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
execute := func(argv []string) ([]Item, error) {
|
||||||
|
item, _, err := Execute("_", argv, nil, nil, "", time.Minute)
|
||||||
|
return item, err
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err := execute([]string{"true"})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
// Exit with error code
|
||||||
|
res, err = execute([]string{"false"})
|
||||||
|
assertNotNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
res, err = execute([]string{"sh", "-c", "exit 22"})
|
||||||
|
assertNotNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
// Timeout
|
||||||
|
res, _, err = Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond)
|
||||||
|
assertNotNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
// Returning items
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "foo"}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Id != "foo" {
|
||||||
|
t.Fatal("jq -cn test failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read from stdin
|
||||||
|
res, _, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute)
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Id != "bar" {
|
||||||
|
t.Fatal("jq -cR test failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set env
|
||||||
|
res, _, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute)
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Id != "baz" {
|
||||||
|
t.Fatal("jq -cn env test failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// With logging on stderr
|
||||||
|
res, err = execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Id != "box" {
|
||||||
|
t.Fatal("stderr test failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Unsupported item field is silently discarded
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
|
||||||
|
// Field with incorrect type fails
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: ["list"]}`})
|
||||||
|
assertNotNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
|
||||||
|
assertNotNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: null}`})
|
||||||
|
assertNotNil(err)
|
||||||
|
assertLen(res, 0)
|
||||||
|
|
||||||
|
// Items with duplicate ids is not a fetch error, but it will fail to update
|
||||||
|
res, err = execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 2)
|
||||||
|
|
||||||
|
// Action keys are detected even with empty values
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Action["hello"] == nil {
|
||||||
|
t.Fatal("missing hello action")
|
||||||
|
}
|
||||||
|
if res[0].Action["goodbye"] != nil {
|
||||||
|
t.Fatal("nonexistent action should key to nil in Action")
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Action["hello"] == nil {
|
||||||
|
t.Fatal("missing hello action")
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Action["hello"] == nil {
|
||||||
|
t.Fatal("missing hello action")
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Action["hello"] == nil {
|
||||||
|
t.Fatal("missing hello action")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read state
|
||||||
|
argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
|
||||||
|
res, _, err = Execute("_", argv, nil, []byte("world"), "", time.Minute)
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if res[0].Title != "Hello world" {
|
||||||
|
t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Write state
|
||||||
|
argv = []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
|
||||||
|
res, newState, err := Execute("_", argv, nil, nil, "", time.Minute)
|
||||||
|
assertNil(err)
|
||||||
|
assertLen(res, 1)
|
||||||
|
if string(newState) != "Hello world" {
|
||||||
|
t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
|
||||||
|
}
|
||||||
|
}
|
||||||
200
core/items.go
Normal file
200
core/items.go
Normal file
@ -0,0 +1,200 @@
|
|||||||
|
package core
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
_ "github.com/mattn/go-sqlite3"
|
||||||
|
)
|
||||||
|
|
||||||
|
func AddItems(db DB, items []Item) error {
|
||||||
|
return db.Transact(func(tx DB) error {
|
||||||
|
stmt, err := tx.Prepare(`
|
||||||
|
insert into items (source, id, active, title, author, body, link, time, action)
|
||||||
|
values (?, ?, ?, ?, ?, ?, ?, ?, jsonb(?))
|
||||||
|
`)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare insert: %v", err)
|
||||||
|
}
|
||||||
|
for _, item := range items {
|
||||||
|
actions, err := json.Marshal(item.Action)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
|
||||||
|
}
|
||||||
|
_, err = stmt.Exec(item.Source, item.Id, true, item.Title, item.Author, item.Body, item.Link, item.Time, actions)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to insert %s/%s: %v", item.Source, item.Id, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set fields in the new item to match the old item where the new item's fields are zero-valued.
|
||||||
|
// This allows sources to omit fields and let an action set them without a later fetch overwriting
|
||||||
|
// the value from the action, e.g. an on-create action archiving a page and setting the link to
|
||||||
|
// point to the archive.
|
||||||
|
func BackfillItem(new *Item, old *Item) {
|
||||||
|
new.Active = old.Active
|
||||||
|
new.Created = old.Created
|
||||||
|
if new.Author == "" {
|
||||||
|
new.Author = old.Author
|
||||||
|
}
|
||||||
|
if new.Body == "" {
|
||||||
|
new.Body = old.Body
|
||||||
|
}
|
||||||
|
if new.Link == "" {
|
||||||
|
new.Link = old.Link
|
||||||
|
}
|
||||||
|
if new.Time == 0 {
|
||||||
|
new.Time = old.Time
|
||||||
|
}
|
||||||
|
if new.Title == "" {
|
||||||
|
new.Title = old.Title
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func UpdateItems(db DB, items []Item) error {
|
||||||
|
return db.Transact(func(tx DB) error {
|
||||||
|
stmt, err := tx.Prepare(`
|
||||||
|
update items
|
||||||
|
set
|
||||||
|
title = ?,
|
||||||
|
author = ?,
|
||||||
|
body = ?,
|
||||||
|
link = ?,
|
||||||
|
time = ?,
|
||||||
|
action = jsonb(?)
|
||||||
|
where source = ?
|
||||||
|
and id = ?
|
||||||
|
`)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
for _, item := range items {
|
||||||
|
actions, err := json.Marshal(item.Action)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
|
||||||
|
}
|
||||||
|
_, err = stmt.Exec(item.Title, item.Author, item.Body, item.Link, item.Time, actions, item.Source, item.Id)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// Deactivate an item, returning its previous active state.
|
||||||
|
func DeactivateItem(db DB, source string, id string) (bool, error) {
|
||||||
|
row := db.QueryRow(`
|
||||||
|
select active
|
||||||
|
from items
|
||||||
|
where source = ? and id = ?
|
||||||
|
`, source, id)
|
||||||
|
var active bool
|
||||||
|
err := row.Scan(&active)
|
||||||
|
if err != nil && errors.Is(err, sql.ErrNoRows) {
|
||||||
|
return false, fmt.Errorf("item %s/%s not found", source, id)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = db.Exec(`
|
||||||
|
update items
|
||||||
|
set active = 0
|
||||||
|
where source = ? and id = ?
|
||||||
|
`, source, id)
|
||||||
|
if err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
return active, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func DeleteItem(db DB, source string, id string) (int64, error) {
|
||||||
|
res, err := db.Exec(`
|
||||||
|
delete from items
|
||||||
|
where source = ?
|
||||||
|
and id = ?
|
||||||
|
`, source, id)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return res.RowsAffected()
|
||||||
|
}
|
||||||
|
|
||||||
|
func getItems(db DB, query string, args ...any) ([]Item, error) {
|
||||||
|
rows, err := db.Query(query, args...)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
var items []Item
|
||||||
|
for rows.Next() {
|
||||||
|
var item Item
|
||||||
|
err = rows.Scan(&item.Source, &item.Id, &item.Created, &item.Active, &item.Title, &item.Author, &item.Body, &item.Link, &item.Time, &item.Action)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
items = append(items, item)
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetItem(db DB, source string, id string) (Item, error) {
|
||||||
|
items, err := getItems(db, `
|
||||||
|
select source, id, created, active, title, author, body, link, time, json(action)
|
||||||
|
from items
|
||||||
|
where source = ?
|
||||||
|
and id = ?
|
||||||
|
order by case when time = 0 then created else time end, id
|
||||||
|
`, source, id)
|
||||||
|
if err != nil {
|
||||||
|
return Item{}, err
|
||||||
|
}
|
||||||
|
if len(items) == 0 {
|
||||||
|
return Item{}, fmt.Errorf("no item in %s with id %s", source, id)
|
||||||
|
}
|
||||||
|
return items[0], nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetAllActiveItems(db DB) ([]Item, error) {
|
||||||
|
return getItems(db, `
|
||||||
|
select
|
||||||
|
source, id, created, active, title, author, body, link, time, json(action)
|
||||||
|
from items
|
||||||
|
where active <> 0
|
||||||
|
order by case when time = 0 then created else time end, id
|
||||||
|
`)
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetAllItems(db DB) ([]Item, error) {
|
||||||
|
return getItems(db, `
|
||||||
|
select
|
||||||
|
source, id, created, active, title, author, body, link, time, json(action)
|
||||||
|
from items
|
||||||
|
order by case when time = 0 then created else time end, id
|
||||||
|
`)
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetActiveItemsForSource(db DB, source string) ([]Item, error) {
|
||||||
|
return getItems(db, `
|
||||||
|
select
|
||||||
|
source, id, created, active, title, author, body, link, time, json(action)
|
||||||
|
from items
|
||||||
|
where
|
||||||
|
source = ?
|
||||||
|
and active <> 0
|
||||||
|
order by case when time = 0 then created else time end, id
|
||||||
|
`, source)
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetAllItemsForSource(db DB, source string) ([]Item, error) {
|
||||||
|
return getItems(db, `
|
||||||
|
select
|
||||||
|
source, id, created, active, title, author, body, link, time, json(action)
|
||||||
|
from items
|
||||||
|
where
|
||||||
|
source = ?
|
||||||
|
order by case when time = 0 then created else time end, id
|
||||||
|
`, source)
|
||||||
|
}
|
||||||
92
core/items_test.go
Normal file
92
core/items_test.go
Normal file
@ -0,0 +1,92 @@
|
|||||||
|
package core
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
_ "github.com/mattn/go-sqlite3"
|
||||||
|
)
|
||||||
|
|
||||||
|
func AssertItemIs(t *testing.T, item Item, expected string) {
|
||||||
|
actual := fmt.Sprintf(
|
||||||
|
"%s/%s/%t/%s/%s/%s/%s/%d",
|
||||||
|
item.Source,
|
||||||
|
item.Id,
|
||||||
|
item.Active,
|
||||||
|
item.Title,
|
||||||
|
item.Author,
|
||||||
|
item.Body,
|
||||||
|
item.Link,
|
||||||
|
item.Time,
|
||||||
|
)
|
||||||
|
if actual != expected {
|
||||||
|
t.Fatalf("expected %s, got %s", expected, actual)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAddItem(t *testing.T) {
|
||||||
|
db := EphemeralDb(t)
|
||||||
|
if err := AddSource(db, "test"); err != nil {
|
||||||
|
t.Fatalf("failed to add source: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := AddItems(db, []Item{
|
||||||
|
{"test", "one", 0, true, "", "", "", "", 0, nil},
|
||||||
|
{"test", "two", 0, true, "title", "author", "body", "link", 123456, nil},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("failed to add items: %v", err)
|
||||||
|
}
|
||||||
|
items, err := GetActiveItemsForSource(db, "test")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to get active items: %v", err)
|
||||||
|
}
|
||||||
|
if len(items) != 2 {
|
||||||
|
t.Fatal("should get two items")
|
||||||
|
}
|
||||||
|
// order is by (time ?? created) so this ordering is correct as long as you don't run it in early 1970
|
||||||
|
AssertItemIs(t, items[0], "test/two/true/title/author/body/link/123456")
|
||||||
|
AssertItemIs(t, items[1], "test/one/true/////0")
|
||||||
|
|
||||||
|
if _, err = DeactivateItem(db, "test", "one"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
items, err = GetActiveItemsForSource(db, "test")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(items) != 1 {
|
||||||
|
t.Fatal("should get one item")
|
||||||
|
}
|
||||||
|
|
||||||
|
items, err = GetAllItemsForSource(db, "test")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(items) != 2 {
|
||||||
|
t.Fatal("should get two items")
|
||||||
|
}
|
||||||
|
|
||||||
|
deleted, err := DeleteItem(db, "test", "one")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if deleted != 1 {
|
||||||
|
t.Fatal("expected one deletion")
|
||||||
|
}
|
||||||
|
|
||||||
|
deleted, err = DeleteItem(db, "test", "one")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if deleted != 0 {
|
||||||
|
t.Fatal("expected no deletion")
|
||||||
|
}
|
||||||
|
|
||||||
|
items, err = GetAllItemsForSource(db, "test")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(items) != 1 {
|
||||||
|
t.Fatal("should get one item")
|
||||||
|
}
|
||||||
|
}
|
||||||
193
core/source.go
193
core/source.go
@ -1,9 +1,6 @@
|
|||||||
package core
|
package core
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"database/sql"
|
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
@ -55,196 +52,6 @@ func DeleteSource(db DB, name string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func AddItems(db DB, items []Item) error {
|
|
||||||
return db.Transact(func(tx DB) error {
|
|
||||||
stmt, err := tx.Prepare(`
|
|
||||||
insert into items (source, id, active, title, author, body, link, time, action)
|
|
||||||
values (?, ?, ?, ?, ?, ?, ?, ?, jsonb(?))
|
|
||||||
`)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to prepare insert: %v", err)
|
|
||||||
}
|
|
||||||
for _, item := range items {
|
|
||||||
actions, err := json.Marshal(item.Action)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
|
|
||||||
}
|
|
||||||
_, err = stmt.Exec(item.Source, item.Id, true, item.Title, item.Author, item.Body, item.Link, item.Time, actions)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to insert %s/%s: %v", item.Source, item.Id, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// Set fields in the new item to match the old item where the new item's fields are zero-valued.
|
|
||||||
// This allows sources to omit fields and let an action set them without a later fetch overwriting
|
|
||||||
// the value from the action, e.g. an on-create action archiving a page and setting the link to
|
|
||||||
// point to the archive.
|
|
||||||
func BackfillItem(new *Item, old *Item) {
|
|
||||||
new.Active = old.Active
|
|
||||||
new.Created = old.Created
|
|
||||||
if new.Author == "" {
|
|
||||||
new.Author = old.Author
|
|
||||||
}
|
|
||||||
if new.Body == "" {
|
|
||||||
new.Body = old.Body
|
|
||||||
}
|
|
||||||
if new.Link == "" {
|
|
||||||
new.Link = old.Link
|
|
||||||
}
|
|
||||||
if new.Time == 0 {
|
|
||||||
new.Time = old.Time
|
|
||||||
}
|
|
||||||
if new.Title == "" {
|
|
||||||
new.Title = old.Title
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func UpdateItems(db DB, items []Item) error {
|
|
||||||
return db.Transact(func(tx DB) error {
|
|
||||||
stmt, err := tx.Prepare(`
|
|
||||||
update items
|
|
||||||
set
|
|
||||||
title = ?,
|
|
||||||
author = ?,
|
|
||||||
body = ?,
|
|
||||||
link = ?,
|
|
||||||
time = ?,
|
|
||||||
action = jsonb(?)
|
|
||||||
where source = ?
|
|
||||||
and id = ?
|
|
||||||
`)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
for _, item := range items {
|
|
||||||
actions, err := json.Marshal(item.Action)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to marshal actions for %s/%s: %v", item.Source, item.Id, err)
|
|
||||||
}
|
|
||||||
_, err = stmt.Exec(item.Title, item.Author, item.Body, item.Link, item.Time, actions, item.Source, item.Id)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// Deactivate an item, returning its previous active state.
|
|
||||||
func DeactivateItem(db DB, source string, id string) (bool, error) {
|
|
||||||
row := db.QueryRow(`
|
|
||||||
select active
|
|
||||||
from items
|
|
||||||
where source = ? and id = ?
|
|
||||||
`, source, id)
|
|
||||||
var active bool
|
|
||||||
err := row.Scan(&active)
|
|
||||||
if err != nil && errors.Is(err, sql.ErrNoRows) {
|
|
||||||
return false, fmt.Errorf("item %s/%s not found", source, id)
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = db.Exec(`
|
|
||||||
update items
|
|
||||||
set active = 0
|
|
||||||
where source = ? and id = ?
|
|
||||||
`, source, id)
|
|
||||||
if err != nil {
|
|
||||||
return false, err
|
|
||||||
}
|
|
||||||
return active, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func DeleteItem(db DB, source string, id string) (int64, error) {
|
|
||||||
res, err := db.Exec(`
|
|
||||||
delete from items
|
|
||||||
where source = ?
|
|
||||||
and id = ?
|
|
||||||
`, source, id)
|
|
||||||
if err != nil {
|
|
||||||
return 0, err
|
|
||||||
}
|
|
||||||
return res.RowsAffected()
|
|
||||||
}
|
|
||||||
|
|
||||||
func getItems(db DB, query string, args ...any) ([]Item, error) {
|
|
||||||
rows, err := db.Query(query, args...)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
var items []Item
|
|
||||||
for rows.Next() {
|
|
||||||
var item Item
|
|
||||||
err = rows.Scan(&item.Source, &item.Id, &item.Created, &item.Active, &item.Title, &item.Author, &item.Body, &item.Link, &item.Time, &item.Action)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
items = append(items, item)
|
|
||||||
}
|
|
||||||
return items, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetItem(db DB, source string, id string) (Item, error) {
|
|
||||||
items, err := getItems(db, `
|
|
||||||
select source, id, created, active, title, author, body, link, time, json(action)
|
|
||||||
from items
|
|
||||||
where source = ?
|
|
||||||
and id = ?
|
|
||||||
order by case when time = 0 then created else time end, id
|
|
||||||
`, source, id)
|
|
||||||
if err != nil {
|
|
||||||
return Item{}, err
|
|
||||||
}
|
|
||||||
if len(items) == 0 {
|
|
||||||
return Item{}, fmt.Errorf("no item in %s with id %s", source, id)
|
|
||||||
}
|
|
||||||
return items[0], nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetAllActiveItems(db DB) ([]Item, error) {
|
|
||||||
return getItems(db, `
|
|
||||||
select
|
|
||||||
source, id, created, active, title, author, body, link, time, json(action)
|
|
||||||
from items
|
|
||||||
where active <> 0
|
|
||||||
order by case when time = 0 then created else time end, id
|
|
||||||
`)
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetAllItems(db DB) ([]Item, error) {
|
|
||||||
return getItems(db, `
|
|
||||||
select
|
|
||||||
source, id, created, active, title, author, body, link, time, json(action)
|
|
||||||
from items
|
|
||||||
order by case when time = 0 then created else time end, id
|
|
||||||
`)
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetActiveItemsForSource(db DB, source string) ([]Item, error) {
|
|
||||||
return getItems(db, `
|
|
||||||
select
|
|
||||||
source, id, created, active, title, author, body, link, time, json(action)
|
|
||||||
from items
|
|
||||||
where
|
|
||||||
source = ?
|
|
||||||
and active <> 0
|
|
||||||
order by case when time = 0 then created else time end, id
|
|
||||||
`, source)
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetAllItemsForSource(db DB, source string) ([]Item, error) {
|
|
||||||
return getItems(db, `
|
|
||||||
select
|
|
||||||
source, id, created, active, title, author, body, link, time, json(action)
|
|
||||||
from items
|
|
||||||
where
|
|
||||||
source = ?
|
|
||||||
order by case when time = 0 then created else time end, id
|
|
||||||
`, source)
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetState(db DB, source string) ([]byte, error) {
|
func GetState(db DB, source string) ([]byte, error) {
|
||||||
row := db.QueryRow("select state from sources where name = ?", source)
|
row := db.QueryRow("select state from sources where name = ?", source)
|
||||||
var state []byte
|
var state []byte
|
||||||
|
|||||||
@ -2,7 +2,6 @@ package core
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"slices"
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
@ -46,90 +45,6 @@ func TestCreateSource(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func AssertItemIs(t *testing.T, item Item, expected string) {
|
|
||||||
actual := fmt.Sprintf(
|
|
||||||
"%s/%s/%t/%s/%s/%s/%s/%d",
|
|
||||||
item.Source,
|
|
||||||
item.Id,
|
|
||||||
item.Active,
|
|
||||||
item.Title,
|
|
||||||
item.Author,
|
|
||||||
item.Body,
|
|
||||||
item.Link,
|
|
||||||
item.Time,
|
|
||||||
)
|
|
||||||
if actual != expected {
|
|
||||||
t.Fatalf("expected %s, got %s", expected, actual)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestAddItem(t *testing.T) {
|
|
||||||
db := EphemeralDb(t)
|
|
||||||
if err := AddSource(db, "test"); err != nil {
|
|
||||||
t.Fatalf("failed to add source: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := AddItems(db, []Item{
|
|
||||||
{"test", "one", 0, true, "", "", "", "", 0, nil},
|
|
||||||
{"test", "two", 0, true, "title", "author", "body", "link", 123456, nil},
|
|
||||||
}); err != nil {
|
|
||||||
t.Fatalf("failed to add items: %v", err)
|
|
||||||
}
|
|
||||||
items, err := GetActiveItemsForSource(db, "test")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("failed to get active items: %v", err)
|
|
||||||
}
|
|
||||||
if len(items) != 2 {
|
|
||||||
t.Fatal("should get two items")
|
|
||||||
}
|
|
||||||
// order is by (time ?? created) so this ordering is correct as long as you don't run it in early 1970
|
|
||||||
AssertItemIs(t, items[0], "test/two/true/title/author/body/link/123456")
|
|
||||||
AssertItemIs(t, items[1], "test/one/true/////0")
|
|
||||||
|
|
||||||
if _, err = DeactivateItem(db, "test", "one"); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
items, err = GetActiveItemsForSource(db, "test")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if len(items) != 1 {
|
|
||||||
t.Fatal("should get one item")
|
|
||||||
}
|
|
||||||
|
|
||||||
items, err = GetAllItemsForSource(db, "test")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if len(items) != 2 {
|
|
||||||
t.Fatal("should get two items")
|
|
||||||
}
|
|
||||||
|
|
||||||
deleted, err := DeleteItem(db, "test", "one")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if deleted != 1 {
|
|
||||||
t.Fatal("expected one deletion")
|
|
||||||
}
|
|
||||||
|
|
||||||
deleted, err = DeleteItem(db, "test", "one")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if deleted != 0 {
|
|
||||||
t.Fatal("expected no deletion")
|
|
||||||
}
|
|
||||||
|
|
||||||
items, err = GetAllItemsForSource(db, "test")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if len(items) != 1 {
|
|
||||||
t.Fatal("should get one item")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestUpdateSourceAddAndDelete(t *testing.T) {
|
func TestUpdateSourceAddAndDelete(t *testing.T) {
|
||||||
db := EphemeralDb(t)
|
db := EphemeralDb(t)
|
||||||
if err := AddSource(db, "test"); err != nil {
|
if err := AddSource(db, "test"); err != nil {
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user