bot: Bot setup and teardown done: Connect w/retry, join/leave, full config, and logging.

internal/models/change.go

@@ -13,13 +13,24 @@ type Change struct {
 	Op      string         `json:"op"`
 	Author  string         `json:"author"`
 	Date    time.Time      `json:"date"`
+	Keys    []ChangeKey    `json:"keys"`
 	Objects []ChangeObject `json:"objects"`
 }
 
+// A ChangeKey describes a model directly or indirectly affected by the change.
+type ChangeKey struct {
+	Model string `json:"model"`
+	ID    string `json:"id"`
+}
+
 // A ChangeObject is an object affected by the change.
 type ChangeObject struct {
 	Data     json.RawMessage `json:"-"`
-	TypeName string          `json:"__typename"`
+	TypeName string          `json:"-"`
+}
+
+type changeObjectHeader struct {
+	TypeName string `json:"__typename"`
 }
 
 // Channel parses the ChangeObject as a channel.
@@ -38,12 +49,15 @@ func (co *ChangeObject) Channel() (Channel, error) {
 func (co *ChangeObject) UnmarshalJSON(b []byte) error {
 	data := make([]byte, len(b))
 	copy(data, b)
-	co.Data = data
 
-	err := json.Unmarshal(data, co)
+	header := changeObjectHeader{}
+	err := json.Unmarshal(data, &header)
 	if err != nil {
 		return err
 	}
 
+	co.Data = data
+	co.TypeName = header.TypeName
+
 	return nil
 }

internal/models/changes/list.go

@@ -0,0 +1,61 @@
+package changes
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"git.aiterp.net/rpdata/logbot3/internal/api"
+	"git.aiterp.net/rpdata/logbot3/internal/models"
+)
+
+// The Filter for List.
+type Filter struct {
+	Keys         []models.ChangeKey `json:"keys,omitempty"`
+	EarliestDate *time.Time         `json:"earliestDate,omitempty"`
+	Limit        int                `json:"limit,omitempty"`
+}
+
+// List lists all changes according to the filter.
+func List(ctx context.Context, filter *Filter) ([]models.Change, error) {
+	data, err := api.Global().Query(ctx, listGQL, map[string]interface{}{"filter": filter}, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	res := listResult{}
+	err = json.Unmarshal(data, &res)
+
+	return res.Changes, err
+}
+
+type listResult struct {
+	Changes []models.Change `json:"changes"`
+}
+
+var listGQL = `
+	query ListChanges($filter:ChangesFilter) {
+		changes(filter:$filter) {
+			id
+			model
+			op
+			author
+			listed
+			date
+			keys {
+				model
+				id
+			}
+			objects {
+				__typename
+				...on Channel {
+					name
+					logged
+					hub
+					eventName
+					locationName
+				}
+			}
+		}
+	}
+`

internal/models/channels/subscribe-logged.go

@@ -2,13 +2,16 @@ package channels
 
 import (
 	"context"
+	"log"
 	"time"
 
 	"git.aiterp.net/rpdata/logbot3/internal/models"
+	"git.aiterp.net/rpdata/logbot3/internal/models/changes"
 )
 
 // SubscribeLogged returns a channel that gets all the channel changes.
 func SubscribeLogged(ctx context.Context) (<-chan models.Channel, error) {
+	earliest := time.Now()
 	channels, err := ListLogged(ctx)
 	if err != nil {
 		return nil, err
@@ -21,8 +24,28 @@ func SubscribeLogged(ctx context.Context) (<-chan models.Channel, error) {
 
 	go func() {
 		for {
-			time.Sleep(time.Second * 20)
+			time.Sleep(time.Second * 5)
 
+			// Get the changes.
+			changes, err := changes.List(ctx, &changes.Filter{EarliestDate: &earliest, Keys: []models.ChangeKey{{Model: "Channel", ID: "*"}}})
+			if ctx.Err() != nil {
+				return
+			} else if err != nil {
+				log.Println("Failed to get chnges:", err)
+				continue
+			}
+
+			// Group changes to the same model.
+			for _, change := range changes {
+				for _, obj := range change.Objects {
+					channel, err := obj.Channel()
+					if err != nil {
+						continue
+					}
+
+					output <- channel
+				}
+			}
 		}
 	}()