fix: csv output handles time and interface values

storage/csv.go

@@ -3,18 +3,22 @@ package storage
 import (
 	"context"
 	"encoding/csv"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
 	"reflect"
 	"sync"
+	"time"
 
 	"github.com/go-pg/pg/v10/orm"
 
 	"github.com/filecoin-project/sentinel-visor/model"
 )
 
+const PostgresTimestampFormat = "2006-01-02T15:04:05.999Z07:00"
+
 var ErrMarshalUnsupportedType = errors.New("cannot marshal unsupported type")
 
 var (
@@ -169,10 +173,29 @@ func (c *CSVBatch) PersistModel(ctx context.Context, m interface{}) error {
 		for i, f := range t.fields {
 			fv := value.FieldByName(f)
 			fk := fv.Kind()
-			if (fk == reflect.Slice || fk == reflect.Map || fk == reflect.Ptr || fk == reflect.Chan || fk == reflect.Func) && fv.IsNil() {
+			if (fk == reflect.Slice || fk == reflect.Map || fk == reflect.Ptr || fk == reflect.Chan || fk == reflect.Func || fk == reflect.Interface) && fv.IsNil() {
 				row[i] = "NULL"
 				continue
 			}
+
+			if fk == reflect.Interface {
+				v, err := json.Marshal(fv.Interface())
+				if err != nil {
+					return err
+				}
+				row[i] = string(v)
+				continue
+			}
+
+			ft := fv.Type()
+
+			// Special formatting for known types
+			if ft.PkgPath() == "time" && ft.Name() == "Time" {
+				v := fv.Interface().(time.Time)
+				row[i] = v.Format(PostgresTimestampFormat)
+				continue
+			}
+
 			row[i] = fmt.Sprint(fv)
 		}
 		c.data[t.name] = append(c.data[t.name], row)

storage/csv_test.go

@@ -2,10 +2,12 @@ package storage
 
 import (
 	"context"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/go-pg/pg/v10"
 	"github.com/stretchr/testify/assert"
@@ -24,6 +26,29 @@ func (tm *TestModel) Persist(ctx context.Context, s model.StorageBatch) error {
 	return s.PersistModel(ctx, tm)
 }
 
+type TimeModel struct {
+	Height    int64     `pg:",pk,notnull,use_zero"`
+	Processed time.Time `pg:",pk,notnull"`
+}
+
+func (tm *TimeModel) Persist(ctx context.Context, s model.StorageBatch) error {
+	return s.PersistModel(ctx, tm)
+}
+
+type InterfaceModel struct {
+	Height int64 `pg:",pk,notnull,use_zero"`
+	Value  interface{}
+}
+
+func (im *InterfaceModel) Persist(ctx context.Context, s model.StorageBatch) error {
+	return s.PersistModel(ctx, im)
+}
+
+type ProcessingError struct {
+	Source string
+	Error  error
+}
+
 func TestCSVTable(t *testing.T) {
 	tm := &TestModel{
 		Height:  42,
@@ -169,3 +194,86 @@ func TestCSVPersistComposite(t *testing.T) {
 			"42\n",
 		string(otherWritten))
 }
+
+func TestCSVPersistTime(t *testing.T) {
+	// use time.Now since the default string value includes the monotonic clock, so we can test it is not present in csv output
+	now := time.Now()
+
+	tm := &TimeModel{
+		Height:    42,
+		Processed: now,
+	}
+
+	dir, err := ioutil.TempDir("", t.Name())
+	require.NoError(t, err)
+
+	defer os.RemoveAll(dir)
+
+	st, err := NewCSVStorage(dir)
+	require.NoError(t, err)
+
+	err = st.PersistBatch(context.Background(), tm)
+	require.NoError(t, err)
+
+	written, err := ioutil.ReadFile(filepath.Join(dir, "time_models.csv"))
+	require.NoError(t, err)
+	assert.EqualValues(t,
+		"height,processed\n"+
+			"42,"+now.Format(PostgresTimestampFormat)+"\n",
+		string(written))
+}
+
+func TestCSVPersistInterfaceNil(t *testing.T) {
+	tm := &InterfaceModel{
+		Height: 42,
+		Value:  nil,
+	}
+
+	dir, err := ioutil.TempDir("", t.Name())
+	require.NoError(t, err)
+
+	defer os.RemoveAll(dir)
+
+	st, err := NewCSVStorage(dir)
+	require.NoError(t, err)
+
+	err = st.PersistBatch(context.Background(), tm)
+	require.NoError(t, err)
+
+	written, err := ioutil.ReadFile(filepath.Join(dir, "interface_models.csv"))
+	require.NoError(t, err)
+	assert.EqualValues(t,
+		"height,value\n"+
+			"42,NULL\n",
+		string(written))
+}
+
+func TestCSVPersistInterfaceValue(t *testing.T) {
+	tm := &InterfaceModel{
+		Height: 42,
+		Value: []*ProcessingError{
+			{
+				Source: "some task",
+				Error:  fmt.Errorf("processing error"),
+			},
+		},
+	}
+
+	dir, err := ioutil.TempDir("", t.Name())
+	require.NoError(t, err)
+
+	defer os.RemoveAll(dir)
+
+	st, err := NewCSVStorage(dir)
+	require.NoError(t, err)
+
+	err = st.PersistBatch(context.Background(), tm)
+	require.NoError(t, err)
+
+	written, err := ioutil.ReadFile(filepath.Join(dir, "interface_models.csv"))
+	require.NoError(t, err)
+	assert.EqualValues(t,
+		"height,value\n"+
+			"42,\"[{\"\"Source\"\":\"\"some task\"\",\"\"Error\"\":{}}]\"\n",
+		string(written))
+}