X-Git-Url: http://git.giacomolli.uk/onechat/blobdiff_plain/47f69638bc96100f2e446f63c7c3689023156e64..49566fe4e40a18434320829c5cb6a6eff1b4ec78:/daemon.go diff --git a/daemon.go b/daemon.go index a4041f1..3a950da 100644 --- a/daemon.go +++ b/daemon.go @@ -27,8 +27,9 @@ const ( // 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. 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. +// telegram.json exists, it also runs the Telegram bot. It also keeps a linked +// WhatsApp device connected. It holds the lock of the token file, so it is the +// only process that refreshes the tokens and opens the WhatsApp session. func daemon(ctx context.Context, out io.Writer) (err error) { dir, err := dataDir() if err != nil { @@ -62,6 +63,16 @@ func daemon(ctx context.Context, out io.Writer) (err error) { err = errors.Join(err, tokens.Close()) }() + whatsappSetDeviceProps() + whatsapp, err := openWhatsApp(ctx, dir, time.Now, log.New(out, "whatsapp: ", log.LstdFlags), whatsappNewClient) + if err != nil { + return err + } + + defer func() { + err = errors.Join(err, whatsapp.Close()) + }() + address, auth, err := modelEndpoint(config, dir) if err != nil { return err @@ -82,6 +93,11 @@ func daemon(ctx context.Context, out io.Writer) (err error) { err = errors.Join(err, memory.Close()) }() + schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), time.Now) + if err != nil { + return err + } + subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), time.Now, subagentNameRandom) defer func() { err = errors.Join(err, subagents.Close()) @@ -106,13 +122,16 @@ func daemon(ctx context.Context, out io.Writer) (err error) { } } - return daemonServe(ctx, listener, telegram, func(notify func(Event)) (*TurnLoop, error) { - return newTurnLoop(agent, memory, subagents, client, time.Now, notify) + return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) { + return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, time.Now, notify) }) } -// DaemonHub holds the connected clients and sends each event to all of them. +// DaemonHub holds the connected clients and sends each event to all of them. It +// runs the WhatsApp commands of the clients on whatsapp. type DaemonHub struct { + whatsapp *WhatsApp + mutex sync.Mutex clients map[*DaemonClient]bool workers sync.WaitGroup @@ -178,7 +197,12 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) var request SocketRequest err := json.Unmarshal(scanner.Bytes(), &request) if err == nil { - err = daemonRunRequest(l, request) + err = daemonRunRequest(l, h.whatsapp, request, func(text string) { + h.Send(client, Event{ + Kind: "notice", + Text: text, + }) + }) } if err != nil { @@ -287,14 +311,18 @@ func daemonListenSocket(path string) (net.Listener, error) { } // daemonServe runs the loop that start makes and serves the clients of 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 daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*TurnLoop, error)) error { +// until ctx ends or the loop fails. It closes listener. It runs whatsapp, which +// logs its batches and notes in the loop and tells all clients its notices, and +// a failure of whatsapp stops the agent. It runs schedules, which log their +// runs in the loop, and a failure of schedules stops the agent too. When bot is +// not nil, it runs the bot too, and a failure of the bot stops the agent. +func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error)) error { ctx, cancel := context.WithCancel(ctx) defer cancel() hub := &DaemonHub{ - clients: make(map[*DaemonClient]bool), + whatsapp: whatsapp, + clients: make(map[*DaemonClient]bool), } notify := hub.Broadcast @@ -311,20 +339,37 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, s } var workers sync.WaitGroup - workers.Add(1) + workers.Add(2) go func() { defer workers.Done() hub.Accept(ctx, listener, l) }() + var whatsappErr error + go func() { + defer workers.Done() + + whatsappErr = whatsapp.Run(ctx, l, notify) + cancel() + }() + + var schedulesErr error + workers.Add(1) + go func() { + defer workers.Done() + + schedulesErr = schedules.Run(ctx, l) + cancel() + }() + var botErr error if bot != nil { workers.Add(1) go func() { defer workers.Done() - botErr = bot.Run(ctx, l) + botErr = bot.Run(ctx, l, whatsapp) cancel() }() } @@ -339,11 +384,12 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, s hub.CloseAll() workers.Wait() - return errors.Join(err, botErr) + return errors.Join(err, whatsappErr, schedulesErr, botErr) } -// daemonRunRequest runs one request of a client. -func daemonRunRequest(l *TurnLoop, request SocketRequest) error { +// daemonRunRequest runs one request of a client. The answers of a WhatsApp +// command go to answer. +func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, answer func(text string)) error { switch request.Type { case "send": if request.Text == "" { @@ -358,6 +404,10 @@ func daemonRunRequest(l *TurnLoop, request SocketRequest) error { return errors.New("no turn runs") } + return nil + case "whatsapp": + whatsapp.Command(request.Text, answer) + return nil case "stop": found, err := l.StopSubagent(request.Text)