diff --git a/app/jobs/webhooks/facebook_delivery_job.rb b/app/jobs/webhooks/facebook_delivery_job.rb
new file mode 100644
index 000000000..e6b7d1dcf
--- /dev/null
+++ b/app/jobs/webhooks/facebook_delivery_job.rb
@@ -0,0 +1,8 @@
+class Webhooks::FacebookDeliveryJob < ApplicationJob
+  queue_as :low
+
+  def perform(message)
+    response = ::Integrations::Facebook::MessageParser.new(message)
+    Integrations::Facebook::DeliveryStatus.new(params: response).perform
+  end
+end
diff --git a/config/initializers/facebook_messenger.rb b/config/initializers/facebook_messenger.rb
index f9f4fae88..354687cbd 100644
--- a/config/initializers/facebook_messenger.rb
+++ b/config/initializers/facebook_messenger.rb
@@ -29,14 +29,13 @@ Rails.application.reloader.to_prepare do
   end
 
   Facebook::Messenger::Bot.on :delivery do |delivery|
-    # delivery.ids       # => 'mid.1457764197618:41d102a3e1ae206a38'
-    # delivery.sender    # => { 'id' => '1008372609250235' }
-    # delivery.recipient # => { 'id' => '2015573629214912' }
-    # delivery.at        # => 2016-04-22 21:30:36 +0200
-    # delivery.seq       # => 37
-    updater = Integrations::Facebook::DeliveryStatus.new(delivery)
-    updater.perform
-    Rails.logger.info "Human was online at #{delivery.at}"
+    Rails.logger.info "Recieved delivery status #{delivery.to_json}"
+    Webhooks::FacebookDeliveryJob.perform_later(delivery.to_json)
+  end
+
+  Facebook::Messenger::Bot.on :read do |read|
+    Rails.logger.info "Recieved read status  #{read.to_json}"
+    Webhooks::FacebookDeliveryJob.perform_later(read.to_json)
   end
 
   Facebook::Messenger::Bot.on :message_echo do |message|
diff --git a/lib/integrations/facebook/delivery_status.rb b/lib/integrations/facebook/delivery_status.rb
index d38ece092..1d6257bba 100644
--- a/lib/integrations/facebook/delivery_status.rb
+++ b/lib/integrations/facebook/delivery_status.rb
@@ -1,32 +1,37 @@
 # frozen_string_literal: true
 
 class Integrations::Facebook::DeliveryStatus
-  def initialize(params)
-    @params = params
-  end
+  pattr_initialize [:params!]
 
   def perform
-    update_message_status
+    return if facebook_channel.blank?
+    return unless conversation
+
+    process_delivery_status if params.delivery_watermark
+    process_read_status if params.read_watermark
   end
 
   private
 
-  def sender_id
-    @params.sender['id']
+  def process_delivery_status
+    timestamp = Time.zone.at(params.delivery_watermark.to_i).to_datetime.utc
+    ::Conversations::UpdateMessageStatusJob.perform_later(conversation.id, timestamp, :delivered)
+  end
+
+  def process_read_status
+    timestamp = Time.zone.at(params.read_watermark.to_i).to_datetime.utc
+    ::Conversations::UpdateMessageStatusJob.perform_later(conversation.id, timestamp, :read)
   end
 
   def contact
-    ::ContactInbox.find_by(source_id: sender_id)&.contact
+    ::ContactInbox.find_by(source_id: params.sender_id)&.contact
   end
 
   def conversation
     @conversation ||= ::Conversation.find_by(contact_id: contact.id) if contact.present?
   end
 
-  def update_message_status
-    return unless conversation
-
-    conversation.contact_last_seen_at = @params.at
-    conversation.save!
+  def facebook_channel
+    @facebook_channel ||= Channel::FacebookPage.find_by(page_id: params.recipient_id)
   end
 end
diff --git a/lib/integrations/facebook/message_parser.rb b/lib/integrations/facebook/message_parser.rb
index 123b521cb..275a46c39 100644
--- a/lib/integrations/facebook/message_parser.rb
+++ b/lib/integrations/facebook/message_parser.rb
@@ -34,6 +34,22 @@ class Integrations::Facebook::MessageParser
     @messaging.dig('message', 'mid')
   end
 
+  def delivery
+    @messaging['delivery']
+  end
+
+  def read
+    @messaging['read']
+  end
+
+  def read_watermark
+    read&.dig('watermark')
+  end
+
+  def delivery_watermark
+    delivery&.dig('watermark')
+  end
+
   def echo?
     @messaging.dig('message', 'is_echo')
   end
diff --git a/spec/factories/facebook_message/incoming_fb_text_message.rb b/spec/factories/facebook_message/incoming_fb_text_message.rb
index 83d516f09..dc75a912c 100644
--- a/spec/factories/facebook_message/incoming_fb_text_message.rb
+++ b/spec/factories/facebook_message/incoming_fb_text_message.rb
@@ -10,4 +10,24 @@ FactoryBot.define do
 
     initialize_with { attributes }
   end
+
+  factory :message_deliveries, class: Hash do
+    messaging do
+      { sender: { id: '3383290475046708' },
+        recipient: { id: '117172741761305' },
+        delivery: { watermark: '1648581633369' } }
+    end
+
+    initialize_with { attributes }
+  end
+
+  factory :message_reads, class: Hash do
+    messaging do
+      { sender: { id: '3383290475046708' },
+        recipient: { id: '117172741761305' },
+        read: { watermark: '1648581633369' } }
+    end
+
+    initialize_with { attributes }
+  end
 end
diff --git a/spec/jobs/webhooks/facebook_delivery_job_spec.rb b/spec/jobs/webhooks/facebook_delivery_job_spec.rb
new file mode 100644
index 000000000..8dc91b9ab
--- /dev/null
+++ b/spec/jobs/webhooks/facebook_delivery_job_spec.rb
@@ -0,0 +1,44 @@
+require 'rails_helper'
+
+RSpec.describe Webhooks::FacebookDeliveryJob do
+  include ActiveJob::TestHelper
+
+  let(:message) { 'test_message' }
+  let(:parsed_message) { instance_double(Integrations::Facebook::MessageParser) }
+  let(:delivery_status) { instance_double(Integrations::Facebook::DeliveryStatus) }
+
+  before do
+    allow(Integrations::Facebook::MessageParser).to receive(:new).with(message).and_return(parsed_message)
+    allow(Integrations::Facebook::DeliveryStatus).to receive(:new).with(params: parsed_message).and_return(delivery_status)
+    allow(delivery_status).to receive(:perform)
+  end
+
+  after do
+    clear_enqueued_jobs
+  end
+
+  describe '#perform_later' do
+    it 'enqueues the job' do
+      expect do
+        described_class.perform_later(message)
+      end.to have_enqueued_job(described_class).with(message).on_queue('low')
+    end
+  end
+
+  describe '#perform' do
+    it 'calls the MessageParser with the correct argument' do
+      expect(Integrations::Facebook::MessageParser).to receive(:new).with(message)
+      described_class.perform_now(message)
+    end
+
+    it 'calls the DeliveryStatus with the correct argument' do
+      expect(Integrations::Facebook::DeliveryStatus).to receive(:new).with(params: parsed_message)
+      described_class.perform_now(message)
+    end
+
+    it 'executes perform on the DeliveryStatus instance' do
+      expect(delivery_status).to receive(:perform)
+      described_class.perform_now(message)
+    end
+  end
+end
diff --git a/spec/lib/integrations/facebook/delivery_status_spec.rb b/spec/lib/integrations/facebook/delivery_status_spec.rb
new file mode 100644
index 000000000..28f1fe982
--- /dev/null
+++ b/spec/lib/integrations/facebook/delivery_status_spec.rb
@@ -0,0 +1,83 @@
+require 'rails_helper'
+
+describe Integrations::Facebook::DeliveryStatus do
+  subject(:message_builder) { described_class.new(message_deliveries, facebook_channel.inbox).perform }
+
+  before do
+    stub_request(:post, /graph\.facebook\.com/)
+  end
+
+  let!(:account) { create(:account) }
+  let!(:facebook_channel) { create(:channel_facebook_page, page_id: '117172741761305') }
+  let!(:message_delivery_object) { build(:message_deliveries).to_json }
+  let!(:message_deliveries) { Integrations::Facebook::MessageParser.new(message_delivery_object) }
+
+  let!(:contact) { create(:contact, account: account) }
+  let(:contact_inbox) { create(:contact_inbox, contact: contact, inbox: facebook_channel.inbox, source_id: '3383290475046708') }
+  let!(:conversation) { create(:conversation, inbox: facebook_channel.inbox, contact: contact, contact_inbox: contact_inbox) }
+
+  let!(:message_read_object) { build(:message_reads).to_json }
+  let!(:message_reads) { Integrations::Facebook::MessageParser.new(message_read_object) }
+  let!(:message1) do
+    create(:message, content: 'facebook message', message_type: 'outgoing', inbox: facebook_channel.inbox, conversation: conversation)
+  end
+  let!(:message2) do
+    create(:message, content: 'facebook message', message_type: 'incoming', inbox: facebook_channel.inbox, conversation: conversation)
+  end
+
+  describe '#perform' do
+    context 'when message_deliveries callback fires' do
+      before do
+        allow(Conversations::UpdateMessageStatusJob).to receive(:perform_later)
+      end
+
+      it 'updates all messages if the status is delivered' do
+        described_class.new(params: message_deliveries).perform
+        expect(Conversations::UpdateMessageStatusJob).to have_received(:perform_later).with(
+          message1.conversation.id,
+          Time.zone.at(message_deliveries.delivery['watermark'].to_i).to_datetime,
+          :delivered
+        )
+      end
+
+      it 'does not update the message status if the message is incoming' do
+        described_class.new(params: message_deliveries).perform
+        expect(message2.reload.status).to eq('sent')
+      end
+
+      it 'does not update the message status if the message was created after the watermark' do
+        message1.update(created_at: 1.day.from_now)
+        message_deliveries.delivery['watermark'] = 1.day.ago.to_i
+        described_class.new(params: message_deliveries).perform
+        expect(message1.reload.status).to eq('sent')
+      end
+    end
+
+    context 'when message_reads callback fires' do
+      before do
+        allow(Conversations::UpdateMessageStatusJob).to receive(:perform_later)
+      end
+
+      it 'updates all messages if the status is read' do
+        described_class.new(params: message_reads).perform
+        expect(Conversations::UpdateMessageStatusJob).to have_received(:perform_later).with(
+          message1.conversation.id,
+          Time.zone.at(message_reads.read['watermark'].to_i).to_datetime,
+          :read
+        )
+      end
+
+      it 'does not update the message status if the message is incoming' do
+        described_class.new(params: message_reads).perform
+        expect(message2.reload.status).to eq('sent')
+      end
+
+      it 'does not update the message status if the message was created after the watermark' do
+        message1.update(created_at: 1.day.from_now)
+        message_reads.read['watermark'] = 1.day.ago.to_i
+        described_class.new(params: message_reads).perform
+        expect(message1.reload.status).to eq('sent')
+      end
+    end
+  end
+end