// 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) } } }