diff --git a/internal/matrix/client.go b/internal/matrix/client.go index 38068fe..7594d9c 100644 --- a/internal/matrix/client.go +++ b/internal/matrix/client.go @@ -29,6 +29,10 @@ import ( // ReactionHandler is called when a reaction is received on a post. type ReactionHandler func(roomID id.RoomID, eventID id.EventID, targetEventID id.EventID, emoji string, userID id.UserID) +// MessageHandler is called when a text message is received in a configured room. +// body is the plain-text content; channel is the configured channel name for the room. +type MessageHandler func(channel string, roomID id.RoomID, eventID id.EventID, sender id.UserID, body string) + // deviceInfo holds persisted device credentials. type deviceInfo struct { AccessToken string `json:"access_token"` @@ -47,6 +51,7 @@ type Client struct { mu sync.RWMutex onReact ReactionHandler + onMessage MessageHandler cancelSync context.CancelFunc } @@ -217,6 +222,23 @@ func (c *Client) SetReactionHandler(fn ReactionHandler) { c.onReact = fn } +// SetMessageHandler registers a callback for text messages in configured channel rooms. +func (c *Client) SetMessageHandler(fn MessageHandler) { + c.mu.Lock() + defer c.mu.Unlock() + c.onMessage = fn +} + +// ChannelForRoom returns the channel name configured for the given room, if any. +func (c *Client) ChannelForRoom(roomID id.RoomID) (string, bool) { + for name, rid := range c.channels { + if rid == roomID { + return name, true + } + } + return "", false +} + // Start begins the Matrix sync loop in the background. func (c *Client) Start(ctx context.Context) { syncer := c.mx.Syncer.(*mautrix.DefaultSyncer) @@ -247,6 +269,27 @@ func (c *Client) Start(ctx context.Context) { } }) + // Listen for text messages in configured channel rooms + syncer.OnEventType(event.EventMessage, func(ctx context.Context, evt *event.Event) { + if evt.Sender == c.userID { + return + } + channel, ok := c.ChannelForRoom(evt.RoomID) + if !ok { + return + } + msg := evt.Content.AsMessage() + if msg == nil || msg.MsgType != event.MsgText { + return + } + c.mu.RLock() + handler := c.onMessage + c.mu.RUnlock() + if handler != nil { + handler(channel, evt.RoomID, evt.ID, evt.Sender, msg.Body) + } + }) + // Auto-join on invite syncer.OnEventType(event.StateMember, func(ctx context.Context, evt *event.Event) { if evt.GetStateKey() == string(c.userID) && evt.Content.AsMember().Membership == event.MembershipInvite { diff --git a/internal/poster/queue.go b/internal/poster/queue.go index 0f9770e..ab6a91c 100644 --- a/internal/poster/queue.go +++ b/internal/poster/queue.go @@ -90,6 +90,26 @@ func (q *Queue) Wait() { <-q.done } +// ForcePost pops the next queued item for the given channel and posts it +// immediately, bypassing min-interval, burst cap, and daily cap. Dedup +// (canonical URL cooldown) still applies. Returns true if an item was sent. +func (q *Queue) ForcePost(channel string) bool { + q.mu.Lock() + items := q.queues[channel] + if len(items) == 0 { + q.mu.Unlock() + return false + } + item := items[0] + q.queues[channel] = items[1:] + q.mu.Unlock() + + slog.Info("force-posting story on demand", + "guid", item.GUID, "channel", channel) + q.postItem(item) + return true +} + func (q *Queue) drain() { q.mu.Lock() channels := make([]string, 0, len(q.queues)) diff --git a/main.go b/main.go index ab1ed3d..98371b5 100644 --- a/main.go +++ b/main.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "os/signal" + "strings" "syscall" "time" @@ -17,6 +18,8 @@ import ( "pete/internal/poster" "pete/internal/scheduler" "pete/internal/storage" + + "maunium.net/go/mautrix/id" ) func main() { @@ -80,6 +83,23 @@ func main() { exp := explainer.New(mx, ollamaClient) poster.SetReactionCallback(exp.Handle) + // Wire !post command: force-publish next queued story for the room's channel. + mx.SetMessageHandler(func(channel string, roomID id.RoomID, eventID id.EventID, sender id.UserID, body string) { + if strings.TrimSpace(body) != "!post" { + return + } + slog.Info("!post requested", "channel", channel, "sender", sender) + if queue.ForcePost(channel) { + return + } + if err := mx.PostThreadedReply(channel, eventID, + "nothing queued for "+channel, + "nothing queued for "+channel+"", + ); err != nil { + slog.Warn("!post: failed to send empty-queue reply", "err", err) + } + }) + // Set up graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel()