Limit the time of each Telegram send call

[onechat] / daemon.go

diff --git a/daemon.go b/daemon.go
index 58bf502b5bba1b31d226f9439ac9038310a253ad..b37e0cf7c29a45fa7d2f3d37a34060b58d4af79b 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -7,6 +7,7 @@ import (
        "errors"
        "fmt"
        "io"
        "errors"
        "fmt"
        "io"

+       "log"

        "net"
        "os"
        "path/filepath"
        "net"
        "os"
        "path/filepath"


@@ -34,6 +35,10 @@ const (
 
        // socketPathBytes is the longest path of a Unix socket on Linux.
        socketPathBytes = 107
 
        // socketPathBytes is the longest path of a Unix socket on Linux.
        socketPathBytes = 107

+
+       // terminalSource is the source of the user messages from the clients of
+       // the socket.
+       terminalSource = "terminal"

 )
 
 // errTokensInUse tells that another onechat process uses the token file.
 )
 
 // errTokensInUse tells that another onechat process uses the token file.


@@ -47,9 +52,9 @@ type clientRequest struct {
 }
 
 // daemon runs "onechat daemon": it owns the root agent and its files and serves
 }
 
 // daemon runs "onechat daemon": it owns the root agent and its files and serves

-// the clients on the Unix socket in the data directory until ctx ends. It holds
-// the lock of the token file, so it is the only process that refreshes the
-// tokens.
+// the clients on the Unix socket in the data directory until ctx ends. When
+// telegram.json exists, it also runs the Telegram bot. It holds the lock of the
+// token file, so it is the only process that refreshes the tokens.

 func daemon(ctx context.Context, out io.Writer) (err error) {
        dir, err := dataDir()
        if err != nil {
 func daemon(ctx context.Context, out io.Writer) (err error) {
        dir, err := dataDir()
        if err != nil {


@@ -69,6 +74,11 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                return err
        }
 
                return err
        }
 

+       botSettings, hasBot, err := loadBotConfig(dir)
+       if err != nil {
+               return err
+       }
+

        tokens, err := lockTokens(dir)
        if err != nil {
                return err
        tokens, err := lockTokens(dir)
        if err != nil {
                return err


@@ -108,14 +118,24 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                return errors.Join(err, listener.Close())
        }
 
                return errors.Join(err, listener.Close())
        }
 

-       return serveAgent(ctx, listener, func(notify func(Event)) (*loop, error) {
+       var telegram *telegramBot
+       if hasBot {
+               telegram = newTelegramBot(newBotAPI(telegramAPI, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, offsetFileName), log.New(out, "telegram: ", log.LstdFlags))
+               _, err = fmt.Fprintln(out, "onechat: the Telegram bot waits for messages of its owner")
+               if err != nil {
+                       return errors.Join(err, listener.Close())
+               }
+       }
+
+       return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) {

                return newLoop(agent, memory, client, time.Now, notify)
        })
 }
 
 // serveAgent runs the loop that start makes and serves the clients of
                return newLoop(agent, memory, client, time.Now, notify)
        })
 }
 
 // serveAgent runs the loop that start makes and serves the clients of

-// listener until ctx ends or the loop fails. It closes listener.
-func serveAgent(ctx context.Context, listener net.Listener, start func(notify func(Event)) (*loop, error)) error {
+// listener until ctx ends or the loop fails. It closes listener. When bot is
+// not nil, it runs the bot too, and a failure of the bot stops the agent.
+func serveAgent(ctx context.Context, listener net.Listener, bot *telegramBot, start func(notify func(Event)) (*loop, error)) error {

        ctx, cancel := context.WithCancel(ctx)
        defer cancel()
 
        ctx, cancel := context.WithCancel(ctx)
        defer cancel()
 


@@ -123,7 +143,15 @@ func serveAgent(ctx context.Context, listener net.Listener, start func(notify fu
                clients: make(map[*hubClient]bool),
        }
 
                clients: make(map[*hubClient]bool),
        }
 

-       l, err := start(hub.Broadcast)
+       notify := hub.Broadcast
+       if bot != nil {
+               notify = func(event Event) {
+                       hub.Broadcast(event)
+                       bot.Notify(event)
+               }
+       }
+
+       l, err := start(notify)

        if err != nil {
                return errors.Join(err, listener.Close())
        }
        if err != nil {
                return errors.Join(err, listener.Close())
        }


@@ -136,17 +164,28 @@ func serveAgent(ctx context.Context, listener net.Listener, start func(notify fu
                hub.Accept(ctx, listener, l)
        }()
 
                hub.Accept(ctx, listener, l)
        }()
 

+       var botErr error
+       if bot != nil {
+               workers.Add(1)
+               go func() {
+                       defer workers.Done()
+
+                       botErr = bot.Run(ctx, l)
+                       cancel()
+               }()
+       }
+

        err = l.Run(ctx)
        err = l.Run(ctx)

+       if errors.Is(err, context.Canceled) {
+               err = nil
+       }
+

        cancel()
        err = errors.Join(err, listener.Close())
        hub.CloseAll()
        workers.Wait()
 
        cancel()
        err = errors.Join(err, listener.Close())
        hub.CloseAll()
        workers.Wait()
 

-       if errors.Is(err, context.Canceled) {
-               return nil
-       }
-
-       return err
+       return errors.Join(err, botErr)

 }
 
 // hub holds the connected clients and sends each event to all of them.
 }
 
 // hub holds the connected clients and sends each event to all of them.


@@ -305,7 +344,7 @@ func runRequest(l *loop, request clientRequest) error {
                        return errors.New("empty message")
                }
 
                        return errors.New("empty message")
                }
 

-               _, err := l.Submit(request.Text)
+               _, err := l.Submit(terminalSource, request.Text)

 
                return err
        case "interrupt":
 
                return err
        case "interrupt":