path: root/api/canvas/canvas.go

package canvas

import (
	"compress/gzip"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strings"
	"time"

	"github.com/gorilla/websocket"

	"penguins.lan/api/types"
	"penguins.lan/database"
)

const (
	Rows  = 512
	Cols  = 512
	Scale = 8

	placeCooldown     = 500 * time.Millisecond
	snapshotThreshold = 500
	defaultColor      = 0xffffff

	writeWait      = 10 * time.Second
	pongWait       = 60 * time.Second
	pingPeriod     = (pongWait * 9) / 10
	maxMessageSize = 256
)

var palette = []int{
	0xffffff, 0xe4e4e4, 0x888888, 0x222222,
	0xffa7d1, 0xe50000, 0xe59500, 0xa06a42,
	0xe5d900, 0x94e044, 0x02be01, 0x00d3dd,
	0x0083c7, 0x0000ea, 0xcf6ee4, 0x820080,
}

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin:     func(r *http.Request) bool { return true },
}

type inbound struct {
	Type string `json:"type"`
	X    int    `json:"x"`
	Y    int    `json:"y"`
	I    int    `json:"i"`
}

type outbound struct {
	Type string `json:"type"`
	X    int    `json:"x"`
	Y    int    `json:"y"`
	C    int    `json:"c"`
	Ms   int64  `json:"ms,omitempty"`
}

func blankCanvas() []byte {
	buf := make([]byte, Rows*Cols*3)
	for i := 0; i < len(buf); i += 3 {
		buf[i] = byte(defaultColor >> 16)
		buf[i+1] = byte((defaultColor >> 8) & 0xff)
		buf[i+2] = byte(defaultColor & 0xff)
	}
	return buf
}

func setPixel(buf []byte, x, y, color int) {
	if x < 0 || x >= Cols || y < 0 || y >= Rows {
		return
	}
	i := (y*Cols + x) * 3
	buf[i] = byte(color >> 16)
	buf[i+1] = byte(color >> 8)
	buf[i+2] = byte(color)
}

func baseline(db *sql.DB) ([]byte, int64, error) {
	snap, err := database.LatestCanvasSnapshot(db)
	if err != nil {
		return nil, 0, err
	}
	if snap != nil && len(snap.Data) == Rows*Cols*3 {
		buf := make([]byte, len(snap.Data))
		copy(buf, snap.Data)
		return buf, snap.ThroughID, nil
	}
	return blankCanvas(), 0, nil
}

func currentCanvas(db *sql.DB) ([]byte, error) {
	buf, through, err := baseline(db)
	if err != nil {
		return nil, err
	}
	pixels, err := database.PixelsAfter(db, through)
	if err != nil {
		return nil, err
	}
	for _, p := range pixels {
		setPixel(buf, p.X, p.Y, p.Color)
	}
	return buf, nil
}

func takeSnapshot(db *sql.DB) error {
	buf, through, err := baseline(db)
	if err != nil {
		return err
	}
	pixels, err := database.PixelsAfter(db, through)
	if err != nil {
		return err
	}
	if len(pixels) == 0 {
		return nil
	}

	maxID := through
	for _, p := range pixels {
		setPixel(buf, p.X, p.Y, p.Color)
		if p.ID > maxID {
			maxID = p.ID
		}
	}
	if err := database.SaveCanvasSnapshot(db, buf, maxID); err != nil {
		return err
	}
	return database.DeleteOldCanvasSnapshots(db)
}

type Hub struct {
	db            *sql.DB
	clients       map[*Client]bool
	register      chan *Client
	unregister    chan *Client
	broadcast     chan []byte
	place         chan placement
	lastPlaced    map[string]time.Time
	sinceSnapshot int
}

type placement struct {
	client  *Client
	x, y, i int
}

func NewHub(db *sql.DB) *Hub {
	return &Hub{
		db:         db,
		clients:    make(map[*Client]bool),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		broadcast:  make(chan []byte),
		place:      make(chan placement),
		lastPlaced: make(map[string]time.Time),
	}
}

func (h *Hub) Run() {
	if _, through, err := baseline(h.db); err == nil {
		if n, err := database.CountPixelsAfter(h.db, through); err == nil {
			h.sinceSnapshot = n
		}
	}

	for {
		select {
		case client := <-h.register:
			h.clients[client] = true
		case client := <-h.unregister:
			if _, ok := h.clients[client]; ok {
				delete(h.clients, client)
				close(client.send)
			}
		case payload := <-h.broadcast:
			h.deliver(payload)
		case p := <-h.place:
			h.handlePlace(p)
		}
	}
}

func (h *Hub) deliver(payload []byte) {
	for client := range h.clients {
		select {
		case client.send <- payload:
		default:
			delete(h.clients, client)
			close(client.send)
		}
	}
}

func (h *Hub) sendTo(client *Client, msg outbound) {
	if payload, err := json.Marshal(msg); err == nil {
		select {
		case client.send <- payload:
		default:
		}
	}
}

func (h *Hub) handlePlace(p placement) {
	if p.x < 0 || p.x >= Cols || p.y < 0 || p.y >= Rows || p.i < 0 || p.i >= len(palette) {
		return
	}

	now := time.Now()
	if last, ok := h.lastPlaced[p.client.userID]; ok {
		if remaining := placeCooldown - now.Sub(last); remaining > 0 {
			h.sendTo(p.client, outbound{Type: "cooldown", Ms: remaining.Milliseconds()})
			return
		}
	}

	color := palette[p.i]
	if _, err := database.SavePixel(h.db, p.client.userID, p.x, p.y, color); err != nil {
		log.Println("canvas: save pixel failed:", err)
		return
	}
	h.lastPlaced[p.client.userID] = now

	if payload, err := json.Marshal(outbound{Type: "place", X: p.x, Y: p.y, C: color}); err == nil {
		h.deliver(payload)
	}

	h.sinceSnapshot++
	if h.sinceSnapshot >= snapshotThreshold {
		if err := takeSnapshot(h.db); err != nil {
			log.Println("canvas: snapshot failed:", err)
		} else {
			h.sinceSnapshot = 0
		}
	}
}

type Client struct {
	hub    *Hub
	conn   *websocket.Conn
	send   chan []byte
	userID string
}

func (c *Client) readPump() {
	defer func() {
		c.hub.unregister <- c
		c.conn.Close()
	}()

	c.conn.SetReadLimit(maxMessageSize)
	c.conn.SetReadDeadline(time.Now().Add(pongWait))
	c.conn.SetPongHandler(func(string) error {
		c.conn.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	for {
		_, raw, err := c.conn.ReadMessage()
		if err != nil {
			break
		}
		var in inbound
		if json.Unmarshal(raw, &in) != nil || in.Type != "place" {
			continue
		}
		c.hub.place <- placement{client: c, x: in.X, y: in.Y, i: in.I}
	}
}

func (c *Client) writePump() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.conn.Close()
	}()

	for {
		select {
		case payload, ok := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil {
				return
			}
		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

type pageConfig struct {
	Rows       int
	Cols       int
	Scale      int
	CooldownMs int64
	Palette    []string
}

func paletteHex() []string {
	out := make([]string, len(palette))
	for i, c := range palette {
		out[i] = fmt.Sprintf("#%06x", c)
	}
	return out
}

func PageContinuation(context *types.RequestContext, req *http.Request, resp http.ResponseWriter) types.ContinuationChain {
	return func(success types.Continuation, _failure types.Continuation) types.ContinuationChain {
		(*context.TemplateData)["Canvas"] = pageConfig{
			Rows:       Rows,
			Cols:       Cols,
			Scale:      Scale,
			CooldownMs: placeCooldown.Milliseconds(),
			Palette:    paletteHex(),
		}
		return success(context, req, resp)
	}
}

func StateContinuation(context *types.RequestContext, req *http.Request, resp http.ResponseWriter) types.ContinuationChain {
	return func(success types.Continuation, _failure types.Continuation) types.ContinuationChain {
		buf, err := currentCanvas(context.DBConn)
		if err != nil {
			log.Println("canvas: state failed:", err)
			resp.WriteHeader(http.StatusInternalServerError)
			return success(context, req, resp)
		}
		resp.Header().Set("Content-Type", "application/octet-stream")
		// mostly-blank canvas, so it gzips down to almost nothing.
		if strings.Contains(req.Header.Get("Accept-Encoding"), "gzip") {
			resp.Header().Set("Content-Encoding", "gzip")
			gz := gzip.NewWriter(resp)
			defer gz.Close()
			gz.Write(buf)
		} else {
			resp.Write(buf)
		}
		return success(context, req, resp)
	}
}

func ServeWSContinuation(hub *Hub) types.Continuation {
	return func(context *types.RequestContext, req *http.Request, resp http.ResponseWriter) types.ContinuationChain {
		return func(success types.Continuation, failure types.Continuation) types.ContinuationChain {
			conn, err := upgrader.Upgrade(resp, req, nil)
			if err != nil {
				log.Println("canvas: upgrade failed", err)
				return failure(context, req, resp)
			}

			client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256), userID: context.User.ID}
			hub.register <- client
			go client.writePump()
			go client.readPump()

			return success(context, req, resp)
		}
	}
}