Part 1 - A simple gRPC service

syntax = "proto3";
import "google/protobuf/timestamp.proto";
import "google/protobuf/struct.proto";

option go_package = "github.com/onehub/protos";
package onehub.v1;

// Artists perform/play/sing songs
message Topic {
  google.protobuf.Timestamp created_at = 1;
  google.protobuf.Timestamp updated_at = 2;

  // ID of the topic
  string id = 3;

  // ID of the user that created this topic
  string creator_id = 4;
  
  // A unique name of the topic that users can use to connect to
  string name = 5;

  // IDs of users in this topic.   Right now no information about
  // their participation is kept.
  repeated string users = 6;
}

/**
 * An individual message in a topic
 */
message Message {
  /**
   * When the message was created on the server.
   */
  google.protobuf.Timestamp created_at = 1;

  /**
   * When the message or its body were last modified (if modifications are
   * possible).
   */
  google.protobuf.Timestamp updated_at = 2;

  /**
   * ID of the message guaranteed to be unique within a topic.
   * Set only by the server and cannot be modified.
   */
  string id = 3;

  /**
   * User sending this message.
   */
  string user_id = 4;

  /**
   * Topic the message is part of.  This is only set by the server
   * and cannot be modified.
   */
  string topic_id = 5;

  /**
   * Content type of the message. Can be like a ContentType http
   * header or something custom like shell/command
   */
  string content_type = 6;

  /**
   * A simple way to just send text.
   */
  string content_text = 7;

  // Raw contents for data stored locally as JSON
  // Note we can have a combination of text, url and data
  // to show different things in the View/UI
  google.protobuf.Struct content_data = 8;
}

syntax = "proto3";
import "google/protobuf/field_mask.proto";

option go_package = "github.com/onehub/protos";
package onehub.v1;

import "onehub/v1/models.proto";

/**
 * Service for operating on topics
 */
service TopicService {
  /**
   * Create a new sesssion
   */
  rpc CreateTopic(CreateTopicRequest) returns (CreateTopicResponse) {
  }

  /**
   * List all topics from a user.
   */
  rpc ListTopics(ListTopicsRequest) returns (ListTopicsResponse) { 
  }

  /**
   * Get a particular topic
   */
  rpc GetTopic(GetTopicRequest) returns (GetTopicResponse) { 
  }

  /**
   * Batch get multiple topics by ID
   */
  rpc GetTopics(GetTopicsRequest) returns (GetTopicsResponse) { 
  }

  /**
   * Delete a particular topic
   */
  rpc DeleteTopic(DeleteTopicRequest) returns (DeleteTopicResponse) { 
  }

  /**
   * Updates specific fields of a topic
   */
  rpc UpdateTopic(UpdateTopicRequest) returns (UpdateTopicResponse) {
  }
}

/**
 * Topic creation request object
 */
message CreateTopicRequest {
  /**
   * Topic being updated
   */
  Topic topic = 1;
}

/**
 * Response of an topic creation.
 */
message CreateTopicResponse {
  /**
   * Topic being created
   */
  Topic topic = 1;
}

/**
 * An topic search request.  For now only paginations params are provided.
 */
message ListTopicsRequest {
  /**
   * Instead of an offset an abstract  "page" key is provided that offers
   * an opaque "pointer" into some offset in a result set.
   */
  string page_key = 1;

  /**
   * Number of results to return.
   */
  int32 page_size = 2;
}

/**
 * Response of a topic search/listing.
 */
message ListTopicsResponse {
  /**
   * The list of topics found as part of this response.
   */
  repeated Topic topics = 1;

  /**
   * The key/pointer string that subsequent List requests should pass to
   * continue the pagination.
   */
  string next_page_key = 2;
}

/**
 * Request to get an topic.
 */
message GetTopicRequest {
  /**
   * ID of the topic to be fetched
   */
  string id = 1;
}

/**
 * Topic get response
 */
message GetTopicResponse {
  Topic topic = 1;
}

/**
 * Request to batch get topics
 */
message GetTopicsRequest {
  /**
   * IDs of the topic to be fetched
   */
  repeated string ids = 1;
}

/**
 * Topic batch-get response
 */
message GetTopicsResponse {
  map<string, Topic> topics = 1;
}

/**
 * Request to delete an topic.
 */
message DeleteTopicRequest {
  /**
   * ID of the topic to be deleted.
   */
  string id = 1;
}

/**
 * Topic deletion response
 */
message DeleteTopicResponse {
}

/**
 * The request for (partially) updating an Topic.
 */
message UpdateTopicRequest {
  /**
   * Topic being updated
   */
  Topic topic = 1;

  /**
   * Mask of fields being updated in this Topic to make partial changes.
   */
  google.protobuf.FieldMask update_mask = 2;

  /**
   * IDs of users to be added to this topic.
   */
  repeated string add_users = 3;

  /**
   * IDs of users to be removed from this topic.
   */
  repeated string remove_users = 4;
}

/**
 * The request for (partially) updating an Topic.
 */
message UpdateTopicResponse {
  /**
   * Topic being updated
   */
  Topic topic = 1;
}

syntax = "proto3";
import "google/protobuf/field_mask.proto";

option go_package = "github.com/onehub/protos";
package onehub.v1;

import "onehub/v1/models.proto";

/**
 * Service for operating on messages
 */
service MessageService {
  /**
   * Create a new sesssion
   */
  rpc CreateMessage(CreateMessageRequest) returns (CreateMessageResponse) {
  }

  /**
   * List all messages in a topic
   */
  rpc ListMessages(ListMessagesRequest) returns (ListMessagesResponse) { 
  }

  /**
   * Get a particular message
   */
  rpc GetMessage(GetMessageRequest) returns (GetMessageResponse) { 
  }

  /**
   * Batch get multiple messages by IDs
   */
  rpc GetMessages(GetMessagesRequest) returns (GetMessagesResponse) { 
  }

  /**
   * Delete a particular message
   */
  rpc DeleteMessage(DeleteMessageRequest) returns (DeleteMessageResponse) { 
  }

  /**
   * Update a message within a topic.
   */
  rpc UpdateMessage(UpdateMessageRequest) returns (UpdateMessageResponse) {
  }
}

/**
 * Message creation request object
 */
message CreateMessageRequest {
  /**
   * Message being updated
   */
  Message message = 1;
}

/**
 * Response of an message creation.
 */
message CreateMessageResponse {
  /**
   * Message being created
   */
  Message message = 1;
}

/**
 * A message listing request.  For now only paginations params are provided.
 */
message ListMessagesRequest {
  /**
   * Instead of an offset an abstract  "page" key is provided that offers
   * an opaque "pointer" into some offset in a result set.
   */
  string page_key = 1;

  /**
   * Number of results to return.
   */
  int32 page_size = 2;

  /**
   * Topic in which messages are to be listed.  Required.
   */
  string topic_id = 3;
}

/**
 * Response of a topic search/listing.
 */
message ListMessagesResponse {
  /**
   * The list of topics found as part of this response.
   */
  repeated Message messages = 1;

  /**
   * The key/pointer string that subsequent List requests should pass to
   * continue the pagination.
   */
  string next_page_key = 2;
}

/**
 * Request to get a single message.
 */
message GetMessageRequest {
  /**
   * ID of the topic to be fetched
   */
  string id = 1;
}

/**
 * Message get response
 */
message GetMessageResponse {
  Message message = 1;
}

/**
 * Request to batch get messages
 */
message GetMessagesRequest {
  /**
   * IDs of the messages to be fetched
   */
  repeated string ids = 1;
}

/**
 * Message batch-get response
 */
message GetMessagesResponse {
  map<string, Message> messages = 1;
}

/**
 * Request to delete an message
 */
message DeleteMessageRequest {
  /**
   * ID of the message to be deleted.
   */
  string id = 1;
}

/**
 * Message deletion response
 */
message DeleteMessageResponse {
}

message UpdateMessageRequest {
  // The message being updated.  The topic ID AND message ID fields *must*
  // be specified in this message object.  How other fields are used is
  // determined by the update_mask parameter enabling partial updates
  Message message = 1;

  // Indicates which fields are being updated
  // If the field_mask is *not* provided then we reject
  // a replace (as required by the standard convention) to prevent
  // full replace in error.  Instead an update_mask of "*" must be passed.
  google.protobuf.FieldMask update_mask = 3;

  // Any fields specified here will be "appended" to instead of being
  // replaced
  google.protobuf.FieldMask append_mask = 4;
}

message UpdateMessageResponse {
  // The updated message
  Message message = 1;
}

# Some vars to detemrine go locations etc
GOROOT=$(which go)
GOPATH=$(HOME)/go
GOBIN=$(GOPATH)/bin

# Evaluates the abs path of the directory where this Makefile resides
SRC_DIR:=$(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))

# Where the protos exist
PROTO_DIR:=$(SRC_DIR)/protos

# where we want to generate server stubs, clients etc
OUT_DIR:=$(SRC_DIR)/gen/go

all: printenv goprotos

goprotos:
	echo "Generating GO bindings"
	rm -Rf $(OUT_DIR) && mkdir -p $(OUT_DIR)
	protoc --go_out=$(OUT_DIR) --go_opt=paths=source_relative          	\
       --go-grpc_out=$(OUT_DIR) --go-grpc_opt=paths=source_relative		\
       --proto_path=$(PROTO_DIR) 																			\
      $(PROTO_DIR)/onehub/v1/*.proto

printenv:
	@echo MAKEFILE_LIST=$(MAKEFILE_LIST)
	@echo SRC_DIR=$(SRC_DIR)
	@echo PROTO_DIR=$(PROTO_DIR)
	@echo OUT_DIR=$(OUT_DIR)
	@echo GOROOT=$(GOROOT)
	@echo GOPATH=$(GOPATH)
	@echo GOBIN=$(GOBIN)

package services

import (
	"fmt"
	"log"
	"sort"
	"time"

	tspb "google.golang.org/protobuf/types/known/timestamppb"
)

type EntityStore[T any] struct {
	IDCount  int
	Entities map[string]*T

	// Getters/Setters for ID
	IDSetter func(entity *T, id string)
	IDGetter func(entity *T) string

	// Getters/Setters for created timestamp
	CreatedAtSetter func(entity *T, ts *tspb.Timestamp)
	CreatedAtGetter func(entity *T) *tspb.Timestamp

	// Getters/Setters for udpated timestamp
	UpdatedAtSetter func(entity *T, ts *tspb.Timestamp)
	UpdatedAtGetter func(entity *T) *tspb.Timestamp
}

func NewEntityStore[T any]() *EntityStore[T] {
	return &EntityStore[T]{
		Entities: make(map[string]*T),
	}
}

func (s *EntityStore[T]) Create(entity *T) *T {
	s.IDCount++
	newid := fmt.Sprintf("%d", s.IDCount)
	s.Entities[newid] = entity
	s.IDSetter(entity, newid)
	s.CreatedAtSetter(entity, tspb.New(time.Now()))
	s.UpdatedAtSetter(entity, tspb.New(time.Now()))
	return entity
}

func (s *EntityStore[T]) Get(id string) *T {
	if entity, ok := s.Entities[id]; ok {
		return entity
	}
	return nil
}

func (s *EntityStore[T]) BatchGet(ids []string) map[string]*T {
	out := make(map[string]*T)
	for _, id := range ids {
		if entity, ok := s.Entities[id]; ok {
			out[id] = entity
		}
	}
	return out
}

// Updates specific fields of an Entity
func (s *EntityStore[T]) Update(entity *T) *T {
	s.UpdatedAtSetter(entity, tspb.New(time.Now()))
	return entity
}

// Deletes an entity from our system.
func (s *EntityStore[T]) Delete(id string) bool {
	_, ok := s.Entities[id]
	if ok {
		delete(s.Entities, id)
	}
	return ok
}

// Finds and retrieves entity matching the particular criteria.
func (s *EntityStore[T]) List(ltfunc func(t1, t2 *T) bool, filterfunc func(t *T) bool) (out []*T) {
	log.Println("E: ", s.Entities)
	for _, ent := range s.Entities {
		if filterfunc == nil || filterfunc(ent) {
			out = append(out, ent)
		}
	}
	// Sort in reverse order of name
	sort.Slice(out, func(idx1, idx2 int) bool {
		ent1 := out[idx1]
		ent2 := out[idx2]
		return ltfunc(ent1, ent2)
	})
	return
}

package services

import (
	"context"
	"log"
	"strings"

	protos "github.com/panyam/onehub/gen/go/onehub/v1"
	tspb "google.golang.org/protobuf/types/known/timestamppb"
)

type TopicService struct {
	protos.UnimplementedTopicServiceServer
	*EntityStore[protos.Topic]
}

func NewTopicService(estore *EntityStore[protos.Topic]) *TopicService {
	if estore == nil {
		estore = NewEntityStore[protos.Topic]()
	}
	estore.IDSetter = func(topic *protos.Topic, id string) { topic.Id = id }
	estore.IDGetter = func(topic *protos.Topic) string { return topic.Id }

	estore.CreatedAtSetter = func(topic *protos.Topic, val *tspb.Timestamp) { topic.CreatedAt = val }
	estore.CreatedAtGetter = func(topic *protos.Topic) *tspb.Timestamp { return topic.CreatedAt }

	estore.UpdatedAtSetter = func(topic *protos.Topic, val *tspb.Timestamp) { topic.UpdatedAt = val }
	estore.UpdatedAtGetter = func(topic *protos.Topic) *tspb.Timestamp { return topic.UpdatedAt }

	return &TopicService{
		EntityStore: estore,
	}
}

// Create a new Topic
func (s *TopicService) CreateTopic(ctx context.Context, req *protos.CreateTopicRequest) (resp *protos.CreateTopicResponse, err error) {
	resp = &protos.CreateTopicResponse{}
	resp.Topic = s.EntityStore.Create(req.Topic)
	return
}

// Get a single topic by id
func (s *TopicService) GetTopic(ctx context.Context, req *protos.GetTopicRequest) (resp *protos.GetTopicResponse, err error) {
	log.Println("Getting Topic by ID: ", req.Id)
	resp = &protos.GetTopicResponse{
		Topic: s.EntityStore.Get(req.Id),
	}
	return
}

// Batch gets multiple topics.
func (s *TopicService) GetTopics(ctx context.Context, req *protos.GetTopicsRequest) (resp *protos.GetTopicsResponse, err error) {
	log.Println("BatchGet for IDs: ", req.Ids)
	resp = &protos.GetTopicsResponse{
		Topics: s.EntityStore.BatchGet(req.Ids),
	}
	return
}

// Updates specific fields of an Topic
func (s *TopicService) UpdateTopic(ctx context.Context, req *protos.UpdateTopicRequest) (resp *protos.UpdateTopicResponse, err error) {
	resp = &protos.UpdateTopicResponse{
		Topic: s.EntityStore.Update(req.Topic),
	}
	return
}

// Deletes an topic from our system.
func (s *TopicService) DeleteTopic(ctx context.Context, req *protos.DeleteTopicRequest) (resp *protos.DeleteTopicResponse, err error) {
	resp = &protos.DeleteTopicResponse{}
	s.EntityStore.Delete(req.Id)
	return
}

// Finds and retrieves topics matching the particular criteria.
func (s *TopicService) ListTopics(ctx context.Context, req *protos.ListTopicsRequest) (resp *protos.ListTopicsResponse, err error) {
	results := s.EntityStore.List(func(s1, s2 *protos.Topic) bool {
		return strings.Compare(s1.Name, s2.Name) < 0
	}, nil)
	log.Println("Found Topics: ", results)
	resp = &protos.ListTopicsResponse{Topics: results}
	return
}

package main

import (
	"flag"
	"log"
	"net"

	"google.golang.org/grpc"

	v1 "github.com/panyam/onehub/gen/go/onehub/v1"
	svc "github.com/panyam/onehub/services"

	// This is needed to enable the use of the grpc_cli tool
	"google.golang.org/grpc/reflection"
)

var (
	addr = flag.String("addr", ":9000", "Address to start the onehub grpc server on.")
)

func startGRPCServer(addr string) {
	// create new gRPC server
	server := grpc.NewServer()
	v1.RegisterTopicServiceServer(server, svc.NewTopicService(nil))
	v1.RegisterMessageServiceServer(server, svc.NewMessageService(nil))
	if l, err := net.Listen("tcp", addr); err != nil {
		log.Fatalf("error in listening on port %s: %v", addr, err)
	} else {
		// the gRPC server
		log.Printf("Starting grpc endpoint on %s:", addr)
		reflection.Register(server)
		if err := server.Serve(l); err != nil {
			log.Fatal("unable to start server", err)
		}
	}
}

func main() {
	flag.Parse()
	startGRPCServer(*addr)
}