diff --git a/amqp/consumer.go b/amqp/consumer.go
index c8b0396..398d53c 100644
--- a/amqp/consumer.go
+++ b/amqp/consumer.go
@@ -194,9 +194,10 @@ func (c *consumer) doDispatch(msg amqplib.Delivery, h *handler, retryCount int32
 	}
 
 	log := logger.WithFields(logrus.Fields{
-		"action":     h.action,
-		"body":       string(msg.Body),
-		"message_id": msg.MessageId,
+		"action":       h.action,
+		"body":         string(msg.Body),
+		"message_id":   msg.MessageId,
+		"published_at": msg.Timestamp.String(),
 	})
 
 	if err == nil {
diff --git a/amqp/producer.go b/amqp/producer.go
index 25ba8fd..9f500c5 100644
--- a/amqp/producer.go
+++ b/amqp/producer.go
@@ -95,7 +95,7 @@ func (p *producer) Publish(action string, data []byte) {
 		Timestamp:    now,
 		Body:         data,
 		Headers: amqp.Table{
-			"x-epoch-milli": now.Unix() / int64(time.Millisecond),
+			"x-epoch-milli": int64(now.UnixNano()/int64(time.Nanosecond)) / int64(time.Millisecond),
 		},
 	})
 }