1 // mmc -- Mattermost client
2 // Copyright (C) 2023-2024 Sergey Matveev <stargrave@stargrave.org>
4 // This program is free software: you can redistribute it and/or modify
5 // it under the terms of the GNU Affero General Public License as
6 // published by the Free Software Foundation, either version 3 of the
9 // This program is distributed in the hope that it will be useful,
10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 // GNU Affero General Public License for more details.
14 // You should have received a copy of the GNU Affero General Public License
15 // along with this program. If not, see <http://www.gnu.org/licenses/>.
40 "github.com/davecgh/go-spew/spew"
41 "github.com/mattermost/mattermost-server/v6/model"
42 "go.cypherpunks.ru/netrc"
43 "go.stargrave.org/mmc"
47 Newwin = flag.String("newwin", "cmd/newwin", "Path to newwin command")
52 func rewriteIfChanged(fn string, data string) {
53 if their, err := os.ReadFile(fn); err != nil ||
54 !bytes.Equal([]byte(data), their) {
55 if err = os.WriteFile(fn, []byte(data), 0o666); err != nil {
61 func mkFifo(pth string) {
62 if _, err := os.Stat(pth); err == nil {
65 if err := syscall.Mkfifo(pth, 0666); err != nil {
71 entrypoint := flag.String("entrypoint", mmc.GetEntrypoint(), "Entrypoint")
72 notifyCmd := flag.String("notify", "cmd/notify", "Path to notification handler")
73 heartbeatCh := flag.String("heartbeat-ch", "town-square", "Channel for heartbeating")
75 log.SetFlags(log.Lshortfile)
76 log.SetOutput(os.Stdout)
77 UmaskCur = syscall.Umask(0)
78 syscall.Umask(UmaskCur)
82 DebugFd, err = os.OpenFile(
83 "debug", os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
90 entrypointURL, err := url.Parse(*entrypoint)
94 login, password := netrc.Find(entrypointURL.Hostname())
95 if login == "" || password == "" {
96 log.Fatalln("no credentials found for:", entrypointURL.Hostname())
98 c := model.NewAPIv4Client(*entrypoint)
99 c.Login(login, password)
100 me, resp, err := c.GetMe("")
103 spew.Fdump(DebugFd, resp)
108 spew.Fdump(DebugFd, me)
110 log.Println("logged in")
112 time.Sleep(mmc.SleepTime)
113 teams, resp, err := c.GetTeamsForUser(me.Id, "")
116 spew.Fdump(DebugFd, resp)
121 spew.Fdump(DebugFd, teams)
125 var updateQueue []string
126 Chans := make(map[string]*model.Channel)
127 time.Sleep(mmc.SleepTime)
128 page, resp, err := c.GetChannelsForTeamForUser(Team.Id, me.Id, false, "")
131 spew.Fdump(DebugFd, resp)
136 spew.Fdump(DebugFd, page)
138 for _, ch := range page {
143 pth := path.Join("chans", strings.ReplaceAll(ch.Name, ".", "_"))
144 updateQueue = append(updateQueue, pth, ch.Id)
145 os.MkdirAll(pth, 0777)
146 rewriteIfChanged(path.Join(pth, "id"), ch.Id+"\n")
147 rewriteIfChanged(path.Join(pth, "info"), fmt.Sprintf(
153 if _, err := os.Stat(path.Join(pth, mmc.OutRec)); err != nil &&
154 errors.Is(err, fs.ErrNotExist) {
155 if _, err = os.OpenFile(
156 path.Join(pth, mmc.OutRec), os.O_WRONLY|os.O_CREATE, 0o666,
162 usersPth := path.Join(pth, "users")
164 go func(ch *model.Channel) {
166 time.Sleep(mmc.SleepTime)
167 fd, err := os.OpenFile(
168 usersPth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
171 log.Println("OpenFile:", usersPth, err)
175 users, resp, err := c.GetUsersInChannel(ch.Id, n, mmc.PerPage, "")
178 spew.Fdump(DebugFd, resp)
180 log.Println("GetUsersInChannel:", err)
185 spew.Fdump(DebugFd, users)
187 for _, u := range users {
188 fmt.Fprintf(fd, "%s\n", u.Username)
190 if len(users) < mmc.PerPage {
198 pth = path.Join(pth, "in")
200 go func(ch *model.Channel) {
202 fd, err := os.OpenFile(pth, os.O_RDONLY, os.FileMode(0666))
206 data, err := io.ReadAll(fd)
211 if _, err = makePost(c, ch.Id, string(data)); err != nil {
212 log.Println("makePost:", err)
218 Users, err := mmc.GetUsers(c, DebugFd)
223 UsersDC := make(map[string]*model.Channel, len(Users))
224 for _, u := range Users {
225 pth := path.Join("users", strings.ReplaceAll(u.Username, ".", "_"))
226 os.MkdirAll(pth, 0777)
228 path.Join(pth, "name"),
229 fmt.Sprintf("%s %s\n", u.FirstName, u.LastName),
231 rewriteIfChanged(path.Join(pth, "email"), u.Email+"\n")
232 rewriteIfChanged(path.Join(pth, "id"), u.Id+"\n")
233 if _, err := os.Stat(path.Join(pth, mmc.OutRec)); err != nil &&
234 errors.Is(err, fs.ErrNotExist) {
235 if _, err = os.OpenFile(
236 path.Join(pth, mmc.OutRec), os.O_WRONLY|os.O_CREATE, 0o666,
242 if fi, err := os.Stat(path.Join(pth, mmc.OutRec)); err == nil && fi.Size() > 0 {
243 time.Sleep(mmc.SleepTime)
244 dc, resp, err := c.CreateDirectChannel(me.Id, u.Id)
247 spew.Fdump(DebugFd, resp)
249 log.Println("CreateDirectChannel:", err)
253 spew.Fdump(DebugFd, dc)
256 updateQueue = append(updateQueue, pth, dc.Id)
259 statusPth := path.Join(pth, "status")
261 go func(u *model.User) {
263 time.Sleep(mmc.SleepTime)
264 fd, err := os.OpenFile(
265 statusPth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
268 log.Println("OpenFile:", statusPth, err)
271 status, resp, err := c.GetUserStatus(u.Id, "")
274 spew.Fdump(DebugFd, resp)
276 log.Println("GetUserStatus:", err)
281 spew.Fdump(DebugFd, status)
283 fmt.Fprintf(fd, "%s\n", status.Status)
288 pth = path.Join(pth, "in")
290 go func(u *model.User) {
291 var dc *model.Channel
293 fd, err := os.OpenFile(pth, os.O_RDONLY, os.FileMode(0666))
297 data, err := io.ReadAll(fd)
304 dc, resp, err = c.CreateDirectChannel(me.Id, u.Id)
307 spew.Fdump(DebugFd, resp)
309 log.Println("CreateDirectChannel:", err)
314 spew.Fdump(DebugFd, dc)
317 if _, err = makePost(c, dc.Id, string(data)); err != nil {
318 log.Println("makePost:", err)
324 UserStatus := make(map[string]string)
325 var UserStatusM sync.RWMutex
327 pth := path.Join("users", "status")
330 time.Sleep(mmc.SleepTime)
331 fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
333 log.Println("OpenFile:", pth, err)
336 agg := make(map[string][]string)
338 for name, status := range UserStatus {
339 agg[status] = append(agg[status], name)
341 UserStatusM.RUnlock()
342 statuses := make([]string, 0, len(agg))
343 for status := range agg {
344 sort.Strings(agg[status])
345 statuses = append(statuses, status)
347 sort.Strings(statuses)
348 for _, status := range statuses {
349 fmt.Fprintln(fd, status+":", strings.Join(agg[status], " "))
355 log.Println("syncing", len(updateQueue)/2, "rooms")
356 for len(updateQueue) > 0 {
357 err := updatePosts(c, Users, updateQueue[0], updateQueue[1])
359 log.Println("updatePosts:", err)
361 updateQueue = updateQueue[2:]
363 log.Println("sync done")
364 if *notifyCmd != "" {
365 exec.Command(*notifyCmd, "sync done").Run()
369 os.MkdirAll("file", 0777)
370 pthGet := path.Join("file", "get")
372 pthOut := path.Join("file", "out")
375 time.Sleep(mmc.SleepTime)
376 fd, err := os.OpenFile(pthGet, os.O_RDONLY, os.FileMode(0666))
380 data, err := io.ReadAll(fd)
385 fileId := strings.TrimRight(string(data), " \n")
386 if len(fileId) == 0 {
389 fi, resp, err := c.GetFileInfo(fileId)
391 spew.Fdump(DebugFd, fi, resp)
401 data, resp, err = c.GetFile(fileId)
404 spew.Fdump(DebugFd, resp)
409 fd, err = os.OpenFile(
410 pthOut, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
416 w := tar.NewWriter(fd)
417 if err = w.WriteHeader(&tar.Header{
418 Format: tar.FormatPAX,
419 Typeflag: tar.TypeReg,
421 Size: int64(len(data)),
423 ModTime: time.Unix(fi.CreateAt/1000, 0),
424 PAXRecords: map[string]string{
426 "MM.MIMEType": fi.MimeType,
433 if _, err = io.Copy(w, bytes.NewReader(data)); err != nil {
446 needsShutdown := make(chan os.Signal)
447 switch entrypointURL.Scheme {
449 entrypointURL.Scheme = "ws"
451 entrypointURL.Scheme = "wss"
453 log.Println("unhandled scheme:", entrypointURL.Scheme)
455 wc, err := model.NewWebSocketClient4(entrypointURL.String(), c.AuthToken)
462 t := time.NewTicker(time.Minute)
466 if wc.ListenError != nil {
467 log.Println("ListenError:", wc.ListenError)
468 needsShutdown <- syscall.SIGTERM
472 if *heartbeatCh != "" {
473 if _, _, err = c.ViewChannel(
475 &model.ChannelView{ChannelId: Chans[*heartbeatCh].Id},
477 log.Println("ChannelView:", err)
480 case <-wc.PingTimeoutChannel:
481 log.Println("PING timeout")
482 needsShutdown <- syscall.SIGTERM
484 case e := <-wc.EventChannel:
485 if e == nil || !e.IsValid() {
489 spew.Fdump(DebugFd, e)
493 if userId, ok := data["user_id"]; ok && userId.(string) != "" {
494 user = Users[userId.(string)]
496 log.Println("unknown user for event:", e)
500 switch eventType := e.EventType(); eventType {
501 case model.WebsocketEventTyping:
502 if *notifyCmd != "" {
503 exec.Command(*notifyCmd, "typing: "+user.Username).Run()
505 case model.WebsocketEventPostEdited,
506 model.WebsocketEventPostDeleted,
507 model.WebsocketEventPosted:
508 chName, ok := data["channel_name"].(string)
513 if err = json.NewDecoder(
514 strings.NewReader(data["post"].(string)),
515 ).Decode(&post); err != nil {
519 switch model.ChannelType(data["channel_type"].(string)) {
520 case model.ChannelTypeDirect:
521 userId := strings.TrimPrefix(chName, me.Id+"__")
522 userId = strings.TrimSuffix(userId, "__"+me.Id)
523 user := Users[userId]
525 log.Println("unknown user:", userId)
528 recipient = path.Join("users", user.Username)
529 case model.ChannelTypeOpen:
531 case model.ChannelTypePrivate:
533 case model.ChannelTypeGroup:
534 recipient = path.Join("chans", chName)
537 recipient, Users, []mmc.Post{{P: &post, E: eventType}},
541 case model.WebsocketEventStatusChange:
542 status := data["status"].(string)
544 UserStatus[user.Username] = status
546 if *notifyCmd != "" {
547 exec.Command(*notifyCmd, fmt.Sprintf(
548 "status: %s -> %s", user.Username, status,
551 case model.WebsocketEventHello:
552 case model.WebsocketEventChannelViewed:
556 case resp := <-wc.ResponseChannel:
557 if resp == nil || !resp.IsValid() {
561 spew.Fdump(DebugFd, resp)
563 statuses := make(map[string]string)
564 for userId, status := range resp.Data {
565 status, ok := status.(string)
569 user := Users[userId]
573 statuses[user.Username] = status
575 if len(statuses) > 0 {
577 for u := range UserStatus {
578 delete(UserStatus, u)
580 for u, status := range statuses {
581 UserStatus[u] = status
589 signal.Notify(needsShutdown, syscall.SIGTERM, syscall.SIGINT)
592 log.Println("finished")