diff --git a/app/assets/javascripts/discourse/models/composer.js.es6 b/app/assets/javascripts/discourse/models/composer.js.es6
index f7fc3d7746e..217f48eb627 100644
--- a/app/assets/javascripts/discourse/models/composer.js.es6
+++ b/app/assets/javascripts/discourse/models/composer.js.es6
@@ -1059,8 +1059,16 @@ const Composer = RestModel.extend({
       data.originalText = this.originalText;
     }
 
-    return Draft.save(this.draftKey, this.draftSequence, data)
+    return Draft.save(
+      this.draftKey,
+      this.draftSequence,
+      data,
+      this.messageBus.clientId
+    )
       .then(result => {
+        if (result.draft_sequence) {
+          this.draftSequence = result.draft_sequence;
+        }
         if (result.conflict_user) {
           this.setProperties({
             draftSaving: false,
@@ -1075,10 +1083,27 @@ const Composer = RestModel.extend({
           });
         }
       })
-      .catch(() => {
+      .catch(e => {
+        let draftStatus;
+        const xhr = e && e.jqXHR;
+
+        if (
+          xhr &&
+          xhr.status === 409 &&
+          xhr.responseJSON &&
+          xhr.responseJSON.errors &&
+          xhr.responseJSON.errors.length
+        ) {
+          const json = e.jqXHR.responseJSON;
+          draftStatus = json.errors[0];
+          if (json.extras && json.extras.description) {
+            bootbox.alert(json.extras.description);
+          }
+        }
+
         this.setProperties({
           draftSaving: false,
-          draftStatus: I18n.t("composer.drafts_offline"),
+          draftStatus: draftStatus || I18n.t("composer.drafts_offline"),
           draftConflictUser: null
         });
       });
diff --git a/app/assets/javascripts/discourse/models/draft.js.es6 b/app/assets/javascripts/discourse/models/draft.js.es6
index abba8511f52..62fefef8c70 100644
--- a/app/assets/javascripts/discourse/models/draft.js.es6
+++ b/app/assets/javascripts/discourse/models/draft.js.es6
@@ -21,11 +21,11 @@ Draft.reopenClass({
     return current;
   },
 
-  save(key, sequence, data) {
+  save(key, sequence, data, clientId) {
     data = typeof data === "string" ? data : JSON.stringify(data);
     return ajax("/draft.json", {
       type: "POST",
-      data: { draft_key: key, sequence, data }
+      data: { draft_key: key, sequence, data, owner: clientId }
     });
   }
 });
diff --git a/app/controllers/draft_controller.rb b/app/controllers/draft_controller.rb
index 95f62c862a2..9537634cd74 100644
--- a/app/controllers/draft_controller.rb
+++ b/app/controllers/draft_controller.rb
@@ -11,7 +11,41 @@ class DraftController < ApplicationController
   end
 
   def update
-    Draft.set(current_user, params[:draft_key], params[:sequence].to_i, params[:data])
+    sequence =
+      begin
+        Draft.set(
+          current_user,
+          params[:draft_key],
+          params[:sequence].to_i,
+          params[:data],
+          params[:owner]
+        )
+      rescue Draft::OutOfSequence
+
+        begin
+          if !Draft.exists?(user_id: current_user.id, draft_key: params[:draft_key])
+            Draft.set(
+              current_user,
+              params[:draft_key],
+              DraftSequence.current(current_user, params[:draft_key]),
+              params[:data],
+              params[:owner]
+            )
+          else
+            raise Draft::OutOfSequence
+          end
+
+        rescue Draft::OutOfSequence
+          render_json_error I18n.t('draft.sequence_conflict_error.title'),
+            status: 409,
+            extras: {
+              description: I18n.t('draft.sequence_conflict_error.description')
+            }
+          return
+        end
+      end
+
+    json = success_json.merge(draft_sequence: sequence)
 
     if data = JSON::parse(params[:data])
       # this is a bit of a kludge we need to remove (all the parsing) too many special cases here
@@ -20,16 +54,21 @@ class DraftController < ApplicationController
         post = Post.find_by(id: data["postId"])
         if post && post.raw != data["originalText"]
           conflict_user = BasicUserSerializer.new(post.last_editor, root: false)
-          return render json: success_json.merge(conflict_user: conflict_user)
+          render json: json.merge(conflict_user: conflict_user)
+          return
         end
       end
     end
 
-    render json: success_json
+    render json: json
   end
 
   def destroy
-    Draft.clear(current_user, params[:draft_key], params[:sequence].to_i)
+    begin
+      Draft.clear(current_user, params[:draft_key], params[:sequence].to_i)
+    rescue Draft::OutOfSequence
+      # nothing really we can do here, if try clearing a draft that is not ours, just skip it.
+    end
     render json: success_json
   end
 
diff --git a/app/models/draft.rb b/app/models/draft.rb
index c31186cf576..610b8ef26dc 100644
--- a/app/models/draft.rb
+++ b/app/models/draft.rb
@@ -5,44 +5,122 @@ class Draft < ActiveRecord::Base
   NEW_PRIVATE_MESSAGE ||= 'new_private_message'
   EXISTING_TOPIC ||= 'topic_'
 
-  def self.set(user, key, sequence, data)
+  class OutOfSequence < StandardError; end
+
+  def self.set(user, key, sequence, data, owner = nil)
     if SiteSetting.backup_drafts_to_pm_length > 0 && SiteSetting.backup_drafts_to_pm_length < data.length
       backup_draft(user, key, sequence, data)
     end
 
-    if d = find_draft(user, key)
-      return if d.sequence > sequence
+    # this is called a lot so we should micro optimize here
+    draft_id, current_owner, current_sequence = DB.query_single(<<~SQL, user_id: user.id, key: key)
+      WITH draft AS (
+        SELECT id, owner FROM drafts
+        WHERE
+          user_id = :user_id AND
+          draft_key = :key
+      ),
+      draft_sequence AS (
+        SELECT sequence
+        FROM draft_sequences
+        WHERE
+          user_id = :user_id AND
+          draft_key = :key
+      )
 
-      DB.exec(<<~SQL, id: d.id, sequence: sequence, data: data)
+      SELECT
+        (SELECT id FROM draft),
+        (SELECT owner FROM draft),
+        (SELECT sequence FROM draft_sequence)
+    SQL
+
+    current_sequence ||= 0
+
+    if draft_id
+      if current_sequence != sequence
+        raise Draft::OutOfSequence
+      end
+
+      if owner && current_owner && current_owner != owner
+        sequence += 1
+
+        DraftSequence.upsert({
+            sequence: sequence,
+            draft_key: key,
+            user_id: user.id,
+          },
+          unique_by: [:user_id, :draft_key]
+        )
+      end
+
+      DB.exec(<<~SQL, id: draft_id, sequence: sequence, data: data, owner: owner || current_owner)
         UPDATE drafts
            SET sequence = :sequence
              , data = :data
              , revisions = revisions + 1
+             , owner = :owner
          WHERE id = :id
       SQL
+
+    elsif sequence != current_sequence
+      raise Draft::OutOfSequence
     else
-      Draft.create!(user_id: user.id, draft_key: key, data: data, sequence: sequence)
+      Draft.create!(
+        user_id: user.id,
+        draft_key: key,
+        data: data,
+        sequence: sequence,
+        owner: owner
+      )
     end
 
-    true
+    sequence
   end
 
   def self.get(user, key, sequence)
-    d = find_draft(user, key)
-    d.data if d && d.sequence == sequence
+
+    opts = {
+      user_id: user.id,
+      draft_key: key,
+      sequence: sequence
+    }
+
+    current_sequence, data, draft_sequence = DB.query_single(<<~SQL, opts)
+      WITH draft AS (
+        SELECT data, sequence
+        FROM drafts
+        WHERE draft_key = :draft_key AND user_id = :user_id
+      ),
+      draft_sequence AS (
+        SELECT sequence
+        FROM draft_sequences
+        WHERE draft_key = :draft_key AND user_id = :user_id
+      )
+      SELECT
+        ( SELECT sequence FROM draft_sequence) ,
+        ( SELECT data FROM draft ),
+        ( SELECT sequence FROM draft )
+    SQL
+
+    current_sequence ||= 0
+
+    if sequence != current_sequence
+      raise Draft::OutOfSequence
+    end
+
+    data if current_sequence == draft_sequence
   end
 
   def self.clear(user, key, sequence)
-    d = find_draft(user, key)
-    d.destroy if d && d.sequence <= sequence
-  end
+    current_sequence = DraftSequence.current(user, key)
 
-  def self.find_draft(user, key)
-    if user.is_a?(User)
-      find_by(user_id: user.id, draft_key: key)
-    else
-      find_by(user_id: user, draft_key: key)
+    # bad caller is a reason to complain
+    if sequence != current_sequence
+      raise Draft::OutOfSequence
     end
+
+    # corrupt data is not a reason not to leave data
+    Draft.where(user_id: user.id, draft_key: key).destroy_all
   end
 
   def self.stream(opts = nil)
diff --git a/config/locales/server.en.yml b/config/locales/server.en.yml
index 4686244be00..969cd73aefc 100644
--- a/config/locales/server.en.yml
+++ b/config/locales/server.en.yml
@@ -871,6 +871,10 @@ en:
       short_description: "Like this post"
       long_form: "liked this"
 
+  draft:
+    sequence_conflict_error:
+      title: "draft error"
+      description: "Draft is being edited in another window. Please reload this page."
   draft_backup:
     pm_title: "Backup Drafts from ongoing topics"
     pm_body: "Topic containing backup drafts"
diff --git a/db/migrate/20191031042212_add_owner_to_drafts.rb b/db/migrate/20191031042212_add_owner_to_drafts.rb
new file mode 100644
index 00000000000..3f33cfda9e9
--- /dev/null
+++ b/db/migrate/20191031042212_add_owner_to_drafts.rb
@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+class AddOwnerToDrafts < ActiveRecord::Migration[6.0]
+  def change
+    add_column :drafts, :owner, :string
+  end
+end
diff --git a/spec/models/draft_spec.rb b/spec/models/draft_spec.rb
index fff8b8d5795..1d6050fdaa9 100644
--- a/spec/models/draft_spec.rb
+++ b/spec/models/draft_spec.rb
@@ -17,15 +17,15 @@ describe Draft do
         random_key: "random"
       }
 
-      Draft.set(user, "new_private_message", 0, draft.to_json)
+      Draft.set(user, "xyz", 0, draft.to_json)
       draft["reply"] = "test" * 100
 
       half_grace = (SiteSetting.editing_grace_period / 2 + 1).seconds
 
       freeze_time half_grace.from_now
-      Draft.set(user, "new_private_message", 77, draft.to_json)
+      Draft.set(user, "xyz", 0, draft.to_json)
 
-      draft_post = BackupDraftPost.find_by(user_id: user.id, key: "new_private_message").post
+      draft_post = BackupDraftPost.find_by(user_id: user.id, key: "xyz").post
 
       expect(draft_post.revisions.count).to eq(0)
 
@@ -33,7 +33,7 @@ describe Draft do
 
       # this should trigger a post revision as 10 minutes have passed
       draft["reply"] = "hello"
-      Draft.set(user, "new_private_message", 77, draft.to_json)
+      Draft.set(user, "xyz", 0, draft.to_json)
 
       draft_topic = BackupDraftTopic.find_by(user_id: user.id)
       expect(draft_topic.topic.posts_count).to eq(2)
@@ -65,11 +65,48 @@ describe Draft do
     expect(Draft.get(user, "test", 0)).to eq nil
   end
 
+  it "should cross check with DraftSequence table" do
+
+    Draft.set(user, "test", 0, "old")
+    expect(Draft.get(user, "test", 0)).to eq "old"
+
+    DraftSequence.next!(user, "test")
+    seq = DraftSequence.next!(user, "test")
+    expect(seq).to eq(2)
+
+    expect do
+      Draft.set(user, "test", seq - 1, "error")
+    end.to raise_error(Draft::OutOfSequence)
+
+    expect do
+      Draft.set(user, "test", seq + 1, "error")
+    end.to raise_error(Draft::OutOfSequence)
+
+    Draft.set(user, "test", seq, "data")
+    expect(Draft.get(user, "test", seq)).to eq "data"
+
+    expect do
+      expect(Draft.get(user, "test", seq - 1)).to eq "data"
+    end.to raise_error(Draft::OutOfSequence)
+
+    expect do
+      expect(Draft.get(user, "test", seq + 1)).to eq "data"
+    end.to raise_error(Draft::OutOfSequence)
+  end
+
   it "should disregard old draft if sequence decreases" do
     Draft.set(user, "test", 0, "data")
+    DraftSequence.next!(user, "test")
     Draft.set(user, "test", 1, "hello")
-    Draft.set(user, "test", 0, "foo")
-    expect(Draft.get(user, "test", 0)).to eq nil
+
+    expect do
+      Draft.set(user, "test", 0, "foo")
+    end.to raise_error(Draft::OutOfSequence)
+
+    expect do
+      Draft.get(user, "test", 0)
+    end.to raise_error(Draft::OutOfSequence)
+
     expect(Draft.get(user, "test", 1)).to eq "hello"
   end
 
@@ -89,7 +126,8 @@ describe Draft do
     Draft.cleanup!
 
     expect(Draft.count).to eq 0
-    Draft.set(Fabricate(:user), Draft::NEW_TOPIC, seq + 1, 'draft')
+
+    Draft.set(Fabricate(:user), Draft::NEW_TOPIC, 0, 'draft')
 
     Draft.cleanup!
 
@@ -160,10 +198,11 @@ describe Draft do
     it 'nukes the post draft when a post is created' do
       user = Fabricate(:user)
       topic = Fabricate(:topic)
-      p = PostCreator.new(user, raw: Fabricate.build(:post).raw, topic_id: topic.id).create
-      Draft.set(p.user, p.topic.draft_key, 0, 'hello')
 
-      PostCreator.new(user, raw: Fabricate.build(:post).raw).create
+      Draft.set(user, topic.draft_key, 0, 'hello')
+
+      p = PostCreator.new(user, raw: Fabricate.build(:post).raw, topic_id: topic.id).create
+
       expect(Draft.get(p.user, p.topic.draft_key, DraftSequence.current(p.user, p.topic.draft_key))).to eq nil
     end
 
@@ -180,7 +219,19 @@ describe Draft do
       Draft.set(u, 'new_topic', 0, 'hello')
       Draft.set(u, 'new_topic', 0, 'goodbye')
 
-      expect(Draft.find_draft(u, 'new_topic').revisions).to eq(2)
+      expect(Draft.find_by(user_id: u.id, draft_key: 'new_topic').revisions).to eq(2)
+    end
+
+    it 'handles owner switching gracefully' do
+      user = Fabricate(:user)
+      draft_seq = Draft.set(user, 'new_topic', 0, 'hello', _owner = 'ABCDEF')
+      expect(draft_seq).to eq(0)
+
+      draft_seq = Draft.set(user, 'new_topic', 0, 'hello world', _owner = 'HIJKL')
+      expect(draft_seq).to eq(1)
+
+      draft_seq = Draft.set(user, 'new_topic', 1, 'hello world', _owner = 'HIJKL')
+      expect(draft_seq).to eq(1)
     end
   end
 end
diff --git a/spec/requests/draft_controller_spec.rb b/spec/requests/draft_controller_spec.rb
index 417f792617b..161980b143e 100644
--- a/spec/requests/draft_controller_spec.rb
+++ b/spec/requests/draft_controller_spec.rb
@@ -26,7 +26,6 @@ describe DraftController do
     post = Fabricate(:post, user: user)
 
     post "/draft.json", params: {
-      username: user.username,
       draft_key: "topic",
       sequence: 0,
       data: {
@@ -39,7 +38,6 @@ describe DraftController do
     expect(JSON.parse(response.body)['conflict_user']).to eq(nil)
 
     post "/draft.json", params: {
-      username: user.username,
       draft_key: "topic",
       sequence: 0,
       data: {
@@ -62,4 +60,97 @@ describe DraftController do
     expect(response.status).to eq(200)
     expect(Draft.get(user, 'xxx', 0)).to eq(nil)
   end
+
+  it 'cant trivially resolve conflicts without interaction' do
+
+    user = sign_in(Fabricate(:user))
+
+    DraftSequence.next!(user, "abc")
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: 0,
+      data: { a: "test" }.to_json,
+      owner: "abcdefg"
+    }
+
+    expect(response.status).to eq(200)
+    json = JSON.parse(response.body)
+    expect(json["draft_sequence"]).to eq(1)
+  end
+
+  it 'has a clean protocol for ownership handover' do
+    user = sign_in(Fabricate(:user))
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: 0,
+      data: { a: "test" }.to_json,
+      owner: "abcdefg"
+    }
+
+    expect(response.status).to eq(200)
+
+    json = JSON.parse(response.body)
+    expect(json["draft_sequence"]).to eq(0)
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: 0,
+      data: { b: "test" }.to_json,
+      owner: "hijklmnop"
+    }
+
+    expect(response.status).to eq(200)
+    json = JSON.parse(response.body)
+    expect(json["draft_sequence"]).to eq(1)
+
+    expect(DraftSequence.current(user, "abc")).to eq(1)
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: 1,
+      data: { c: "test" }.to_json,
+      owner: "hijklmnop"
+    }
+
+    expect(response.status).to eq(200)
+    json = JSON.parse(response.body)
+    expect(json["draft_sequence"]).to eq(1)
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: 1,
+      data: { c: "test" }.to_json,
+      owner: "abc"
+    }
+
+    expect(response.status).to eq(200)
+    json = JSON.parse(response.body)
+    expect(json["draft_sequence"]).to eq(2)
+  end
+
+  it 'raises an error for out-of-sequence draft setting' do
+
+    user = sign_in(Fabricate(:user))
+    seq = DraftSequence.next!(user, "abc")
+    Draft.set(user, "abc", seq, { b: "test" }.to_json)
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: seq - 1,
+      data: { a: "test" }.to_json
+    }
+
+    expect(response.status).to eq(409)
+
+    post "/draft.json", params: {
+      draft_key: "abc",
+      sequence: seq + 1,
+      data: { a: "test" }.to_json
+    }
+
+    expect(response.status).to eq(409)
+
+  end
 end
diff --git a/spec/services/user_destroyer_spec.rb b/spec/services/user_destroyer_spec.rb
index ed11feb0116..02f5e506f7d 100644
--- a/spec/services/user_destroyer_spec.rb
+++ b/spec/services/user_destroyer_spec.rb
@@ -127,7 +127,7 @@ describe UserDestroyer do
     end
 
     context "with a draft" do
-      let!(:draft) { Draft.set(user, 'test', 1, 'test') }
+      let!(:draft) { Draft.set(user, 'test', 0, 'test') }
 
       it "removed the draft" do
         UserDestroyer.new(admin).destroy(user)