underbbs/adapter/misskey.go

package adapter

import (
	"fmt"
	. "forge.lightcrystal.systems/lightcrystal/underbbs/models"
	"github.com/yitsushi/go-misskey"
	mkcore "github.com/yitsushi/go-misskey/core"
	mkm "github.com/yitsushi/go-misskey/models"
	n "github.com/yitsushi/go-misskey/services/notes"
	tl "github.com/yitsushi/go-misskey/services/notes/timeline"
	users "github.com/yitsushi/go-misskey/services/users"
	"strings"
	"sync"
	"time"
)

type MisskeyAdapter struct {
	data     chan SocketData
	nickname string
	server   string
	apiKey   string

	mk *misskey.Client

	// unlike the mastodon client, we have to manage combining resources
	// from different API calls instead of streaming them in a single channel

	cache map[string]time.Time
	mtx   sync.RWMutex

	stop chan bool
}

func (self *MisskeyAdapter) Name() string {
	return self.nickname
}

func (self *MisskeyAdapter) Init(settings Settings, data chan SocketData) error {
	fmt.Println("initializing misskey adapter")

	self.nickname = settings.Nickname
	self.server = *settings.Server
	self.apiKey = *settings.ApiKey
	self.data = data

	fmt.Println("getting ready to initialize internal client")

	client, err := misskey.NewClientWithOptions(
		misskey.WithAPIToken(self.apiKey),
		misskey.WithBaseURL("https", self.server, ""),
	)

	if err != nil {
		fmt.Println(err.Error())
		return err
	}
	fmt.Println("misskey client initialized")
	self.mk = client

	self.cache = make(map[string]time.Time)

	return nil
}

func (self *MisskeyAdapter) Subscribe(filter string) []error {
	// misskey streaming API is undocumented....
	// we could try to reverse engineer it by directly connecting to the websocket???
	// alternatively, we can poll timelines, mentions, etc with a cancellation channel,
	// keep a cache of IDs in memory, and send new objects on the data channel

	// TODO: decode the filter so we can send data to the mk services

	// same as in masto, we will want to close and reopen the stop channel
	if self.stop != nil {
		close(self.stop)
	}
	self.stop = make(chan bool)

	// in the background, continuously read data from these API endpoints
	// if they are newer than the cache, convert them to UnderBBS objects
	// and send them on the data channel

	go self.poll()

	return nil
}

func (self *MisskeyAdapter) poll() {

	var latest *time.Time

	notesService := self.mk.Notes()
	timelineService := notesService.Timeline()

	for {
		select {
		case _, ok := <-self.stop:
			if !ok {
				return
			}
		default:

			// TODO: we have to actually decode and pass our filter criteria

			// probe for new notes
			probenote, err := timelineService.Get(tl.GetRequest{
				Limit: 1,
			})
			if err == nil && len(probenote) > 0 && self.isNew(probenote[0]) {
				if latest == nil {
					latest = &probenote[0].CreatedAt
					// this is the first fetch of notes, we can just grab them
					notes, err := timelineService.Get(tl.GetRequest{
						Limit: 100,
					})
					// if latest is nil also get mentions history
					mentions, merr := notesService.Mentions(n.MentionsRequest{
						Limit: 100,
					})
					if err != nil {
						fmt.Println(err.Error())
					}
					if merr != nil {
						fmt.Println(merr.Error())
					}

					// check the cache for everything we just collected
					// if anything is newer or as of yet not in the cache, add it
					// and convert it to a SocketData implementation before sending on data channel
					for _, n := range notes {
						msg := self.toMessageIfNew(n)
						if msg != nil {
							self.data <- msg
						}
					}
					for _, n := range mentions {
						msg := self.toMessageIfNew(n)
						if msg != nil {
							self.data <- msg
						}
					}

				} else {
					for {
						// get notes since latest, until the probe
						notes, err := timelineService.Get(tl.GetRequest{
							SinceDate: uint64(latest.Unix()),
							UntilDate: uint64(probenote[0].CreatedAt.Unix()),
							Limit:     100,
						})
						if err != nil {
							fmt.Println(err.Error())
						}
						for _, n := range notes {
							msg := self.toMessageIfNew(n)
							if msg == nil {
								latest = &probenote[0].CreatedAt
								break
							}
							self.data <- msg
						}
						if *latest == probenote[0].CreatedAt {
							break
						}
					}
				}
			}

			time.Sleep(5 * time.Second)
		}
	}
}

func (self *MisskeyAdapter) isNew(n mkm.Note) bool {
	self.mtx.RLock()
	timestamp, exists := self.cache[n.ID]
	self.mtx.RUnlock()
	return !exists || timestamp.Before(n.CreatedAt)
}

func (self *MisskeyAdapter) toMessageIfNew(n mkm.Note) *Message {
	return self.toMessage(n, false)
}

func (self *MisskeyAdapter) toMessage(n mkm.Note, bustCache bool) *Message {
	self.mtx.RLock()
	timestamp, exists := self.cache[n.ID]
	self.mtx.RUnlock()
	if bustCache || !exists || timestamp.Before(n.CreatedAt) {
		host := mkcore.StringValue(n.User.Host)
		authorId := ""
		if host != "" {
			authorId = fmt.Sprintf("@%s@%s", n.User.Username, host)
		} else {
			authorId = fmt.Sprintf("@%s", n.User.Username)
		}

		self.mtx.Lock()
		self.cache[n.ID] = n.CreatedAt
		self.mtx.Unlock()
		msg := Message{
			Datagram: Datagram{
				Id:       n.ID,
				Uri:      n.URI,
				Protocol: "misskey",
				Adapter:  self.nickname,
				Type:     "message",
				Created:  n.CreatedAt.UnixMilli(),
			},

			Author:      authorId,
			Content:     n.Text,
			Attachments: []Attachment{},
			Visibility:  n.Visibility,
			ReplyTo:     n.ReplyID,
			ReplyCount:  int(n.RepliesCount),
			Replies:     []string{},
		}

		for _, f := range n.Files {
			msg.Attachments = append(msg.Attachments, Attachment{
				Src:      f.URL,
				ThumbSrc: f.ThumbnailURL,
				Size:     f.Size,
				Desc:     f.Comment,
				Created:  f.CreatedAt.UnixMilli(),
			})
		}
		return &msg
	}
	return nil
}

func (self *MisskeyAdapter) toAuthor(usr mkm.User) *Author {
	fmt.Println("converting author: " + usr.ID)
	host := mkcore.StringValue(usr.Host)
	authorId := ""
	if host != "" {
		authorId = fmt.Sprintf("@%s@%s", usr.Username, host)
	} else {
		authorId = fmt.Sprintf("@%s", usr.Username)
	}

	var updated *int64 = nil
	if usr.UpdatedAt != nil {
		updatedTmp := usr.UpdatedAt.UnixMilli()
		updated = &updatedTmp
	}

	author := Author{
		Datagram: Datagram{
			Id:       authorId,
			Uri:      mkcore.StringValue(usr.URL),
			Protocol: "misskey",
			Adapter:  self.nickname,
			Type:     "author",
			Created:  usr.CreatedAt.UnixMilli(),
			Updated:  updated,
		},
		Name:        usr.Name,
		ProfilePic:  usr.AvatarURL,
		ProfileData: usr.Description,
	}

	return &author
}

func (self *MisskeyAdapter) Fetch(etype, ids []string) error {
	for _, id := range ids {
		switch etype {
		case "message":
			data, err := self.mk.Notes().Show(id)
			if err != nil {
				return err
			} else {
				msg := self.toMessage(data, true)
				if msg != nil {
					self.data <- msg
				}
			}
		case "children":
			data, err := self.mk.Notes().Children(n.ChildrenRequest{
				NoteID: id,
				Limit:  100,
			})
			if err != nil {
				return err
			} else {
				for _, n := range data {
					msg := self.toMessage(n, true)
					if msg != nil {
						self.data <- msg
					}
				}
			}
		case "convoy":
			data, err := self.mk.Notes().Conversation(n.ConversationRequest{
				NoteID: id,
				Limit:  100,
			})
			if err != nil {
				return err
			} else {
				for _, n := range data {
					msg := self.toMessage(n, true)
					if msg != nil {
						self.data <- msg
					}
				}
			}
		case "author":
			user := ""
			host := ""
			idParts := strings.Split(id, "@")
			user = idParts[1]
			if len(idParts) == 3 {
				host = idParts[2]
			}

			var hostPtr *string = nil
			if len(host) > 0 {
				hostPtr = &host
			}

			// fmt.Printf("attempting user resolution: @%s@%s\n", user, host)
			data, err := self.mk.Users().Show(users.ShowRequest{
				Username: &user,
				Host:     hostPtr,
			})
			if err != nil {
				return err
			} else {
				a := self.toAuthor(data)
				if a != nil {
					self.data <- a
				}
			}

		}
	}
	return nil
}

func (self *MisskeyAdapter) Do(action string) error {
	return nil
}

func (self *MisskeyAdapter) DefaultSubscriptionFilter() string {
	return ""
}