(fix): add error handling

- stream error response
- remove update service

go.mod

@@ -4,7 +4,7 @@ go 1.21.2
 
 require (
 	github.com/caarlos0/env/v10 v10.0.0
-	github.com/kameshsampath/demo-protos/golang/todo v0.0.0-20231204144625-d1bdccfdd066
+	github.com/kameshsampath/demo-protos/golang/todo v0.0.0-20231206022715-1afc44aab8bf
 	github.com/twmb/franz-go v1.15.2
 	go.uber.org/zap v1.26.0
 	google.golang.org/grpc v1.59.0

go.sum

@@ -8,8 +8,8 @@ github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/kameshsampath/demo-protos/golang/todo v0.0.0-20231204144625-d1bdccfdd066 h1:+w2gew3Vnb/jqt27hi4LceCrw1L5EMDTs8qRV/eBZkU=
-github.com/kameshsampath/demo-protos/golang/todo v0.0.0-20231204144625-d1bdccfdd066/go.mod h1:TrV0a8JgZQi15yOxOJu7bb/tUWJTsH+lRx43avhu7vg=
+github.com/kameshsampath/demo-protos/golang/todo v0.0.0-20231206022715-1afc44aab8bf h1:N1GHZei2VLPorTbjQT7BoPlAOH3SVdEvaLi48WoISAo=
+github.com/kameshsampath/demo-protos/golang/todo v0.0.0-20231206022715-1afc44aab8bf/go.mod h1:xzsq1AM6feGclH2bbMUEH6volbpJbUKPya/4mWht3Kk=
 github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I=
 github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=

internal/impl/todo_impl.go

@@ -34,16 +34,6 @@ func (s *Server) AddTodo(ctx context.Context, req *todo.TodoAddRequest) (*todo.T
 	return marshallAndSend(ctx, s.client, task)
 }
 
-// UpdateTodo implements todo.TodoServer.
-func (s *Server) UpdateTodo(ctx context.Context, req *todo.UpdateTodoStatusRequest) (*todo.TodoResponse, error) {
-	task := &todo.Task{
-		Title:       req.Task.Title,
-		Description: req.Task.Description,
-		Completed:   req.Task.Completed,
-	}
-	return marshallAndSend(ctx, s.client, task)
-}
-
 // TodoList implements todo.TodoServer.
 func (s *Server) TodoList(empty *emptypb.Empty, stream todo.Todo_TodoListServer) error {
 	ch := make(chan result)
@@ -56,26 +46,41 @@ func (s *Server) TodoList(empty *emptypb.Empty, stream todo.Todo_TodoListServer)
 		case r := <-ch:
 			{
 				if errs := r.errors; len(errs) > 0 {
-					log.Errorln("Errors")
-					//TODO send back errors
+					var errors = make([]*todo.Error, len(errs))
 					for _, err := range errs {
-						log.Errorw("Error Details",
+						log.Debugf("Error Details",
 							"Topic", err.Topic,
 							"Partition", err.Partition,
 							"Error", err.Err,
 						)
+						errors = append(errors, &todo.Error{
+							Topic:     err.Topic,
+							Partition: err.Partition,
+							Message:   err.Err.Error(),
+						})
 					}
+					stream.Send(&todo.TodoListResponse{
+						Response: &todo.TodoListResponse_Errors{Errors: &todo.Errors{
+							Error: errors,
+						}},
+					})
 				}
 				b := r.record.Value
 				task := new(todo.Task)
 				if err := proto.Unmarshal(b, task); err != nil {
-					//TODO send back errors
-					log.Error(err)
+					//Skip Sending invalid data, just log the error
+					log.Errorw("Error marshalling task",
+						"Data", string(b),
+						"Error", err.Error())
 				} else {
-					stream.Send(&todo.TodoResponse{
-						Task:      task,
-						Partition: r.record.Partition,
-						Offset:    r.record.Offset,
+					stream.Send(&todo.TodoListResponse{
+						Response: &todo.TodoListResponse_Todo{
+							Todo: &todo.TodoResponse{
+								Task:      task,
+								Partition: r.record.Partition,
+								Offset:    r.record.Offset,
+							},
+						},
 					})
 				}
 			}