Initial commit

1 files changed, 373 insertions, 0 deletions

diff --git a/api/canvas/canvas.go b/api/canvas/canvas.go
new file mode 100644
index 0000000..717d924
--- /dev/null
+++ b/api/canvas/canvas.go
@@ -0,0 +1,373 @@
+// Package canvas is a single shared r/place-style grid. The append-only pixel
+// log in the database is the source of truth; periodic binary snapshots let a
+// new client load the current state without replaying all of history.
+package canvas
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"penguins.lan/api/types"
+	"penguins.lan/database"
+)
+
+const (
+	Rows  = 64
+	Cols  = 64
+	Scale = 8
+
+	placeCooldown     = 2 * time.Second
+	snapshotThreshold = 50
+	defaultColor      = 0xffffff
+
+	writeWait      = 10 * time.Second
+	pongWait       = 60 * time.Second
+	pingPeriod     = (pongWait * 9) / 10
+	maxMessageSize = 256
+)
+
+// r/place 2017 palette.
+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"`
+}
+
+// ---- snapshot / state ------------------------------------------------------
+
+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)
+}
+
+// baseline returns the latest snapshot's bytes and through-id, or a blank canvas.
+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)
+}
+
+// ---- hub -------------------------------------------------------------------
+
+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
+		}
+	}
+}
+
+// ---- client ----------------------------------------------------------------
+
+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
+			}
+		}
+	}
+}
+
+// ---- continuations ---------------------------------------------------------
+
+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")
+		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)
+		}
+	}
+}