Specify time extracted

Author: lawrencejones

tap/output.go

@@ -32,6 +32,8 @@ type Output struct {
 	Schema *Schema `json:"schema,omitempty"`
 	// Record is a record from the stream, if Type == "RECORD".
 	Record map[string]any `json:"record,omitempty"`
+	// TimeExtracted is the time that this record was extracted, if Type == "RECORD".
+	TimeExtracted string `json:"time_extracted,omitempty"`
 	// KeyProperties is a list of strings indicating which properties make up the primary
 	// key for this stream.
 	//

tap/tap.go

@@ -17,8 +17,8 @@ func Run(ctx context.Context, logger kitlog.Logger, ol *OutputLogger, cl *client
 			return err
 		}
 
-		start := time.Now()
-		logger.Log("msg", "loading records", "start", start.Format(time.RFC3339))
+		timeExtracted := time.Now().Format(time.RFC3339)
+		logger.Log("msg", "loading records", "time_extracted", timeExtracted)
 		records, err := stream.GetRecords(ctx, logger, cl)
 		if err != nil {
 			return err
@@ -27,8 +27,10 @@ func Run(ctx context.Context, logger kitlog.Logger, ol *OutputLogger, cl *client
 		logger.Log("msg", "outputting records", "count", len(records))
 		for _, record := range records {
 			op := &Output{
-				Type:   OutputTypeRecord,
-				Record: record,
+				Type:          OutputTypeRecord,
+				Stream:        name,
+				Record:        record,
+				TimeExtracted: timeExtracted,
 			}
 			if err := ol.Log(op); err != nil {
 				return err