From 7336ea2b4740c88e4ce40eadde620dd26a6f102a Mon Sep 17 00:00:00 2001 From: Gisle Aune Date: Wed, 14 Nov 2018 21:01:29 +0100 Subject: [PATCH] bot: Bot setup and teardown done: Connect w/retry, join/leave, full config, and logging. --- .gitignore | 4 + internal/api/client.go | 2 +- internal/bot/bot.go | 140 +++++++++++-------- internal/bot/handler.go | 43 +++++- internal/config/config.go | 10 +- internal/models/change.go | 20 ++- internal/models/changes/list.go | 61 ++++++++ internal/models/channels/subscribe-logged.go | 25 +++- main.go | 7 +- 9 files changed, 236 insertions(+), 76 deletions(-) create mode 100644 internal/models/changes/list.go diff --git a/.gitignore b/.gitignore index d287cd3..9a21749 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,5 @@ +# Binaries debug +debug.exe +logbot3 +logbot3.exe \ No newline at end of file diff --git a/internal/api/client.go b/internal/api/client.go index e187b94..26f3e1a 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -83,7 +83,7 @@ func (client *Client) Query(ctx context.Context, query string, variables map[str if res.StatusCode >= 500 { stuff, _ := ioutil.ReadAll(res.Body) fmt.Println(string(stuff)) - return nil, errors.New("Internal srever error") + return nil, errors.New("Internal server error") } // Parse response diff --git a/internal/bot/bot.go b/internal/bot/bot.go index 7d40ed2..1a79643 100644 --- a/internal/bot/bot.go +++ b/internal/bot/bot.go @@ -3,10 +3,10 @@ package bot import ( "context" "log" - "strings" "time" "git.aiterp.net/gisle/irc" + "git.aiterp.net/rpdata/logbot3/internal/models" "git.aiterp.net/rpdata/logbot3/internal/models/channels" ) @@ -14,16 +14,19 @@ var botKey = "git.aiterp.net/rpdata/logbot.Bot.key" // The Bot is the IRC client. type Bot struct { - commandChannel string - client *irc.Client - ctx context.Context - ctxCancel context.CancelFunc + client *irc.Client + ctx context.Context + ctxCancel context.CancelFunc + loopCtx context.Context + loopCancel context.CancelFunc } // New creates a new Bot. -func New(ctx context.Context, nick string, alternatives []string) *Bot { +func New(ctx context.Context, nick string, alternatives []string, user string, realName string) *Bot { client := irc.New(ctx, irc.Config{ Nick: nick, + User: user, + RealName: realName, Alternatives: alternatives, SendRate: 2, SkipSSLVerification: false, @@ -40,83 +43,98 @@ func New(ctx context.Context, nick string, alternatives []string) *Bot { // Connect connects the bot to the IRC server. This will disconnect already // established connections. -func (bot *Bot) Connect(server string, ssl bool) error { +func (bot *Bot) Connect(server string, ssl bool, maxRetries int) (err error) { if bot.ctxCancel != nil { bot.ctxCancel() } bot.ctx, bot.ctxCancel = context.WithCancel(bot.client.Context()) - return bot.client.Connect(server, ssl) + retries := 0 + for maxRetries == 0 || retries < maxRetries { + err = bot.client.Connect(server, ssl) + if err != nil { + log.Println("Connect failed:", err.Error()) + if maxRetries > 0 && retries < maxRetries { + retries++ + log.Printf("Retrying in 5s (Retry %d/%d)", retries, maxRetries) + } else { + log.Println("Retrying in 5s (No retry limit)") + } + + time.Sleep(time.Second * 10) + continue + } + + return nil + } + + return err } func (bot *Bot) loop() { - ticker := time.NewTicker(time.Second * 10) - defer ticker.Stop() + bot.loopCtx, bot.loopCancel = context.WithCancel(bot.ctx) - log.Println("Client ready.") + channelChanges, err := channels.SubscribeLogged(bot.ctx) + if err != nil { + log.Println("Failed to get channel changes:", err) + return + } for { select { - case <-ticker.C: + case channel := <-channelChanges: { - bot.syncChannels() + channels := []models.Channel{channel} + deadline := time.After(time.Second * 1) + buffering := true + for buffering { + select { + case channel := <-channelChanges: + channels = append(channels, channel) + case <-deadline: + buffering = false + } + } + + decisions := make(map[string]bool) + for _, channel := range channels { + decisions[channel.Name] = channel.Logged + } + joins := make([]string, 0, len(decisions)) + parts := make([]string, 0, len(decisions)) + for channelName, logged := range decisions { + if logged { + if bot.client.Channel(channelName) != nil { + continue + } + joins = append(joins, channelName) + } else { + if bot.client.Channel(channelName) == nil { + continue + } + parts = append(parts, channelName) + } + } + + if len(joins) > 0 { + bot.client.Join(joins...) + } + if len(parts) > 0 { + bot.client.Part(parts...) + } } - case <-bot.ctx.Done(): + case <-bot.loopCtx.Done(): { - log.Println("Spinning down bot main loop.") + log.Println("Spinning down bot loop.") return } } } } -func (bot *Bot) syncChannels() { - channels, err := channels.ListOpen(bot.ctx) - if err != nil { - log.Println("Failed to update channel-list:", err) - return - } - - names := make([]string, 0, len(channels)) - joins := make([]string, 0, len(channels)) - leaves := make([]string, 0, len(names)) - - // Add new channels to join list. - for _, channel := range channels { - name := channel.Name - names = append(names, name) - - if bot.client.Channel(name) == nil { - joins = append(joins, name) - } - } - - // Add no longer logged channels to leave list. -LeaveLoop: - for _, channel := range bot.client.Channels() { - for _, name := range names { - if strings.ToLower(channel.Name()) == strings.ToLower(name) { - continue LeaveLoop - } - } - } - - // Join channels. - if len(joins) > 0 { - log.Println("Joining", strings.Join(joins, ", ")) - bot.client.SendQueuedf("JOIN %s", strings.Join(joins, ",")) - } - - // leave channels. - if len(leaves) > 0 { - log.Println("Leaving", strings.Join(leaves, ", ")) - bot.client.SendQueuedf("PART %s :Channel removed.", strings.Join(leaves, ",")) - } -} - func (bot *Bot) stopLoop() { - if bot.ctxCancel != nil { - bot.ctxCancel() + if bot.loopCancel != nil { + bot.loopCancel() } } diff --git a/internal/bot/handler.go b/internal/bot/handler.go index 0184443..6cc1fd2 100644 --- a/internal/bot/handler.go +++ b/internal/bot/handler.go @@ -5,6 +5,7 @@ import ( "strings" "git.aiterp.net/gisle/irc" + "git.aiterp.net/rpdata/logbot3/internal/config" ) func handler(event *irc.Event, client *irc.Client) { @@ -13,18 +14,52 @@ func handler(event *irc.Event, client *irc.Client) { return } - if event.Kind() == "packet" && len(event.Verb()) == 3 { - log.Printf("(%s) %s %s\n", event.Verb(), strings.Join(event.Args[1:], " "), event.Text) - } - switch event.Name() { + // Handle bot loop case "hook.ready": { + log.Println("Client is ready!") go bot.loop() } case "client.disconnect": { + log.Println("Client disconnected!") bot.stopLoop() + + conf := config.Get().Server + go bot.Connect(conf.Address, conf.SSL, 0) + } + + // Log joins and leaves + case "packet.join": + { + if client.Nick() == event.Nick { + // TODO: Open channel handler + } + + log.Println("(JOIN)", event.Nick, "joined", event.Arg(0)) + } + case "packet.part": + { + if client.Nick() == event.Nick { + // TODO: Close channel handler + } + + log.Println("(PART)", event.Nick, "left", event.Arg(0)) + } + case "packet.quit": + { + log.Println("(QUIT)", event.Nick, "quit") + } + + // Log initial numerics for debugging's sake + case "packet.001", "packet.002", "packet.003", "packet.251", "packet.255", "packet.265", "packet.266", "packet.250", "packet.375", "packet.372", "packet.376": + { + log.Printf("(%s) %s\n", event.Verb(), event.Text) + } + case "packet.005", "packet.254": + { + log.Printf("(%s) %s %s\n", event.Verb(), strings.Join(event.Args, " "), event.Text) } } } diff --git a/internal/config/config.go b/internal/config/config.go index e2b2891..382a8aa 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -10,10 +10,14 @@ import ( // Config represents the logbot's configuration. type Config struct { Bot struct { - Nicks []string `json:"nicks"` - Server string `json:"server"` - CommandChannel string `json:"commandChannel"` + Nicks []string `json:"nicks"` + User string `json:"user"` + RealName string `json:"realName"` } `json:"bot"` + Server struct { + Address string `json:"server"` + SSL bool `json:"ssl"` + } API struct { Endpoint string `json:"endpoint"` Username string `json:"username"` diff --git a/internal/models/change.go b/internal/models/change.go index 6361401..dc4628d 100644 --- a/internal/models/change.go +++ b/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 } diff --git a/internal/models/changes/list.go b/internal/models/changes/list.go new file mode 100644 index 0000000..7703f64 --- /dev/null +++ b/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 + } + } + } + } +` diff --git a/internal/models/channels/subscribe-logged.go b/internal/models/channels/subscribe-logged.go index 1debb08..e3f9575 100644 --- a/internal/models/channels/subscribe-logged.go +++ b/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 + } + } } }() diff --git a/main.go b/main.go index d529874..445e164 100644 --- a/main.go +++ b/main.go @@ -25,10 +25,10 @@ func main() { os.Exit(1) } - bot := bot.New(context.Background(), conf.Bot.Nicks[0], conf.Bot.Nicks[1:]) - err = bot.Connect(conf.Bot.Server, false) + bot := bot.New(context.Background(), conf.Bot.Nicks[0], conf.Bot.Nicks[1:], conf.Bot.User, conf.Bot.RealName) + err = bot.Connect(conf.Server.Address, conf.Server.SSL, 3) if err != nil { - log.Println("Warning:", err) + log.Println("IRC server cannot be reached:", err) os.Exit(1) } @@ -36,6 +36,7 @@ func main() { interrupt := make(chan os.Signal, 1) signal.Notify(interrupt, os.Interrupt) signal.Notify(interrupt, syscall.SIGTERM) + <-interrupt log.Println("Interrupt received, stopping...") }