Khoi's code samples

canvas.js

// This is a React component that render sheet-like table using canvas in my latest project.
// Only visible rows and columns would be renderred. This solution give blazing fast performance
// and minimal memory footprint.

import React, { useRef, useEffect, useMemo } from "react";
import _ from "lodash";

import { renderGrid } from "./grid";
import { getRenderableContentCols, renderContentCells } from "./rows-content";
import { renderPKCells } from "./primary-key-columns";
import { renderHeaderCells } from "./header";
import styles from "./canvas.module.sass";

export const Canvas = ({
  width,
  height,
  cols,
  rows,
  activeRowInd,
  activeColInd,
  selectedRows,
  selectedCols,
  scrollLeft,
  scrollTop,
  canvasStyles,
}) => {
  const dpr = window.devicePixelRatio || 1;
  const el = useRef(null);

  const pkCols = useMemo(() => _.filter(cols, (col) => col.numRow || col.pk), [
    cols,
  ]);
  const numRows = useMemo(
    () => Math.min(rows.length, Math.ceil(height / canvasStyles.rowHeight)),
    [height, canvasStyles.rowHeight, rows]
  );
  const selectedRowsMap = useMemo(() => {
    if (!selectedRows) return {};
    const m = {};
    for (let ind of selectedRows) {
      m[ind] = true;
    }
    return m;
  }, [selectedRows]);
  const selectedColsMap = useMemo(() => {
    if (!selectedCols) return {};
    const m = {};
    for (let ind of selectedCols) {
      m[ind] = true;
    }
    return m;
  }, [selectedCols]);

  useEffect(() => {
    const canvas = el.current;
    const ctx = canvas.getContext("2d");
    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.fillStyle = "white";
    ctx.fillRect(0, 0, width, height);
    ctx.resetTransform();

    if (height <= 0 || rows.length === 0) {
      return;
    }

    const contentTop =
      canvasStyles.headerHeight - (scrollTop % canvasStyles.rowHeight);
    const startInd = Math.floor(scrollTop / canvasStyles.rowHeight);
    const {
      renderableContentColIndices,
      contentLeft,
    } = getRenderableContentCols({
      cols,
      bbWidth: width,
      scrollLeft,
    });
    const startColInd = renderableContentColIndices[0];
    const renderableContentCols = _.map(
      renderableContentColIndices,
      (ind) => cols[ind]
    );
    const renderableRows = _.compact(
      _.map(
        _.range(startInd, _.min([startInd + numRows, rows.length])),
        (ind) => rows[ind]
      )
    );

    ctx.scale(dpr, dpr);
    ctx.translate(contentLeft, contentTop);
    renderGrid({
      styles: canvasStyles,
      cols: renderableContentCols,
      numRows,
      startInd,
      startColInd,
      activeColInd,
      activeRowInd,
      selectedColsMap,
      selectedRowsMap,
      gridWidth: width - contentLeft,
      ctx,
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(contentLeft, contentTop);
    renderContentCells({
      styles: {
        ...canvasStyles,
        font: canvasStyles.cellFont,
      },
      ctx,
      cols: renderableContentCols,
      rows: renderableRows,
      cellStartInd: renderableContentColIndices[0],
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(0, contentTop);
    renderGrid({
      styles: canvasStyles,
      cols: pkCols,
      numRows,
      startInd,
      startColInd: 1,
      activeColInd,
      activeRowInd,
      selectedColsMap,
      gridWidth: width,
      selectedRowsMap,
      ctx,
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(0, contentTop);
    renderPKCells({
      styles: {
        ...canvasStyles,
        font: canvasStyles.cellFont,
        textColor: canvasStyles.pkTextColor,
      },
      ctx,
      cols: pkCols,
      rows: renderableRows,
      startInd,
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(contentLeft, 0);
    renderGrid({
      styles: {
        ...canvasStyles,
        rowHeight: canvasStyles.headerHeight,
      },
      cols: renderableContentCols,
      numRows: 1,
      startInd,
      startColInd,
      activeColInd,
      activeRowInd,
      selectedColsMap,
      gridWidth: width - contentLeft,
      selectedRowsMap,
      ctx,
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(contentLeft, 0);
    renderHeaderCells({
      styles: canvasStyles,
      ctx,
      cols: renderableContentCols,
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(0, 0);
    renderGrid({
      styles: {
        ...canvasStyles,
        rowHeight: canvasStyles.headerHeight,
      },
      cols: pkCols,
      numRows: 1,
      startInd,
      startColInd: 1,
      activeColInd,
      activeRowInd,
      selectedColsMap,
      gridWidth: width,
      selectedRowsMap,
      ctx,
    });

    ctx.resetTransform();
    ctx.scale(dpr, dpr);
    ctx.translate(0, 0);
    renderHeaderCells({
      styles: {
        ...canvasStyles,
        headerTextColor: canvasStyles.pkTextColor,
      },
      ctx,
      cols: pkCols,
    });
  }, [
    width,
    height,
    scrollLeft,
    scrollTop,
    cols,
    rows,
    activeRowInd,
    activeColInd,
    selectedRowsMap,
    selectedColsMap,
  ]);

  return (
    <canvas
      width={width * dpr}
      height={height * dpr}
      className={styles.canvas}
      ref={el}
    />
  );
};

caster.go

// This package expose a server that can broadcast job update events in real time to user's browser
// I use server-sent events for maximum efficiency and minimal latency.

package caster

import (
	"context"
	"encoding/json"

	"github.com/go-pg/pg/v9"
	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"

	"app/internal/pkg/auth"
	"app/internal/pkg/loggers"
	"app/internal/pkg/rbac"
	"app/internal/pkg/workflow"
)

var statusString = map[workflow.JobStatus]string{
	workflow.JOB_STATUS_PENDING:   "pending",
	workflow.JOB_STATUS_RUNNING:   "running",
	workflow.JOB_STATUS_SUCCEED:   "succeed",
	workflow.JOB_STATUS_FAILED:    "failed",
	workflow.JOB_STATUS_CANCELLED: "cancelled",
}

type casterContextKey string

type Caster struct {
	e                 *echo.Echo
	db                *pg.DB
	jobsBroker        *Broker
	jobsChan          <-chan *pg.Notification
	jobDefsInsertChan <-chan *pg.Notification
	jobDefsDeleteChan <-chan *pg.Notification
	jobsTotalVec      *prometheus.GaugeVec
	jobDurationVec    *prometheus.HistogramVec
}

func NewCaster(db *pg.DB, rbacEnforcer *rbac.Enforcer, jwtMan auth.JWTManager) (*Caster, error) {
	err := workflow.CreateJobUpdateNotification(db)
	if err != nil {
		return nil, err
	}
	err = workflow.CreateJobDefNotifications(db)
	if err != nil {
		return nil, err
	}

	c := &Caster{
		db:         db,
		jobsBroker: NewBroker(),
	}

	collector, err := c.register(prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Name: "jobs_total",
		Help: "Total count of jobs",
	}, []string{"org_id", "status"}))
	if err != nil {
		return nil, err
	}
	c.jobsTotalVec = collector.(*prometheus.GaugeVec)

	collector, err = c.register(prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Name:    "job_duration_seconds",
		Help:    "Duration of job in seconds",
		Buckets: []float64{10, 60, 5 * 60, 30 * 60},
	}, []string{"org_id"}))
	if err != nil {
		return nil, err
	}
	c.jobDurationVec = collector.(*prometheus.HistogramVec)

	e := echo.New()
	skipper := middleware.Skipper(func(c echo.Context) bool {
		return c.Path() == "/metrics"
	})
	loggerConf := middleware.DefaultLoggerConfig
	loggerConf.Skipper = skipper
	e.Use(middleware.LoggerWithConfig(loggerConf))
	e.Use(middleware.Recover())
	e.Use(auth.AuthorizationMiddleware(db, jwtMan, skipper))
	e.Use(auth.RBACAuthMiddleware(db, rbacEnforcer, skipper))
	e.Use(c.SetTopic)
	e.GET("/caster/watch-jobs", echo.WrapHandler(c.jobsBroker))
	e.GET("/metrics", echo.WrapHandler(promhttp.Handler()))
	c.e = e

	return c, nil
}

func (c *Caster) watchJobsTopic(pid string) string {
	return "watchJobs/" + pid
}

func (c *Caster) register(collector prometheus.Collector) (prometheus.Collector, error) {
	if err := prometheus.Register(collector); err != nil {
		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
			collector = are.ExistingCollector
		} else {
			return nil, err
		}
	}
	return collector, nil
}

func (caster *Caster) SetTopic(next echo.HandlerFunc) echo.HandlerFunc {
	return func(c echo.Context) error {
		if c.Path() == "/metrics" {
			return next(c)
		}
		pid := c.Get("project_id").(string)
		req := c.Request()
		topic := caster.watchJobsTopic(pid)
		c.SetRequest(req.WithContext(context.WithValue(req.Context(), casterContextKey("Topic"), topic)))
		return next(c)
	}
}

func (c *Caster) HandleJobsNotification() {
	for {
		select {
		case n := <-c.jobsChan:
			j := &workflow.Job{}
			err := json.Unmarshal([]byte(n.Payload), j)
			if err != nil {
				loggers.Error.Printf(err.Error())
				continue
			}
			c.jobsTotalVec.With(prometheus.Labels{
				"org_id": j.OrgID,
				"status": statusString[j.Status],
			}).Inc()
			if j.OldStatus != workflow.JOB_STATUS_UNSPECIFIED {
				c.jobsTotalVec.With(prometheus.Labels{
					"org_id": j.OrgID,
					"status": statusString[j.OldStatus],
				}).Dec()
			}
			if !j.FinishTime.IsZero() && !j.StartTime.IsZero() {
				duration := j.FinishTime.Sub(j.StartTime)
				c.jobDurationVec.With(prometheus.Labels{
					"org_id": j.OrgID,
				}).Observe(duration.Seconds())
			}
			b, err := json.Marshal(j)
			if err != nil {
				loggers.Error.Printf(err.Error())
				continue
			}
			evt := Event{
				Topic: c.watchJobsTopic(j.ProjectID),
				Event: "job_update",
				Data:  string(b),
			}
			c.jobsBroker.Notifier <- evt
		case n := <-c.jobDefsInsertChan:
			jd := &workflow.JobDef{}
			err := json.Unmarshal([]byte(n.Payload), jd)
			if err != nil {
				loggers.Error.Printf(err.Error())
				continue
			}
			b, err := json.Marshal(jd)
			if err != nil {
				loggers.Error.Printf(err.Error())
				continue
			}
			evt := Event{
				Topic: c.watchJobsTopic(jd.ProjectID),
				Event: "job_def_insert",
				Data:  string(b),
			}
			c.jobsBroker.Notifier <- evt
		case n := <-c.jobDefsDeleteChan:
			jd := &workflow.JobDef{}
			err := json.Unmarshal([]byte(n.Payload), jd)
			if err != nil {
				loggers.Error.Printf(err.Error())
				continue
			}
			b, err := json.Marshal(jd)
			if err != nil {
				loggers.Error.Printf(err.Error())
				continue
			}
			evt := Event{
				Topic: c.watchJobsTopic(jd.ProjectID),
				Event: "job_def_delete",
				Data:  string(b),
			}
			c.jobsBroker.Notifier <- evt
		}
	}
}

func (c *Caster) Start() error {
	ln := workflow.ListenForJobsUpdate(c.db)
	c.jobsChan = ln.Channel()
	ln = workflow.ListenForJobDefsInsert(c.db)
	c.jobDefsInsertChan = ln.Channel()
	ln = workflow.ListenForJobDefsDelete(c.db)
	c.jobDefsDeleteChan = ln.Channel()
	go c.HandleJobsNotification()
	return c.e.Start("0.0.0.0:80")
}